Import CI ipv6-noproxy-cidr test from RHEL
This is from RHEL distgit tests/python-requests, directory Sanity/ipv6-noproxy-cidr. The commit hash this was copied from: 050c2979f27cc70506ddf669f9e0803a7b67c2d5 Only the necessary files were kept.
This commit is contained in:
parent
54447e6554
commit
b5aa2db244
7
tests/ipv6-noproxy-cidr/main.fmf
Normal file
7
tests/ipv6-noproxy-cidr/main.fmf
Normal file
@ -0,0 +1,7 @@
|
|||||||
|
require:
|
||||||
|
- nmap-ncat
|
||||||
|
- python3
|
||||||
|
- python3-requests
|
||||||
|
|
||||||
|
test: "${PYTHON:-python3} repro.py"
|
||||||
|
tty: "true"
|
74
tests/ipv6-noproxy-cidr/repro.py
Normal file
74
tests/ipv6-noproxy-cidr/repro.py
Normal file
@ -0,0 +1,74 @@
|
|||||||
|
import subprocess
|
||||||
|
import requests
|
||||||
|
import os
|
||||||
|
import time
|
||||||
|
import atexit
|
||||||
|
|
||||||
|
subprocesses = []
|
||||||
|
|
||||||
|
python = os.environ.get("PYTHON","python3")
|
||||||
|
server_url = os.environ.get("SERVER_URL", None)
|
||||||
|
testing_no_proxy = os.environ.get("TESTING_NO_PROXY", None)
|
||||||
|
|
||||||
|
# Clean up
|
||||||
|
@atexit.register
|
||||||
|
def cleanup():
|
||||||
|
print("Cleaning up subprocesses")
|
||||||
|
for process in subprocesses:
|
||||||
|
process.terminate()
|
||||||
|
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
for process in subprocesses:
|
||||||
|
process.wait()
|
||||||
|
|
||||||
|
# Part one, assert that everything works
|
||||||
|
if server_url is None:
|
||||||
|
print("starting server")
|
||||||
|
p = subprocess.Popen([python, "server.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
|
subprocesses.append(p)
|
||||||
|
time.sleep(1)
|
||||||
|
server_url = "http://[::1]:8888"
|
||||||
|
no_proxy = "::1/128"
|
||||||
|
else:
|
||||||
|
print(f"using provided {server_url} as server")
|
||||||
|
print(f"using provided {testing_no_proxy} as NO_PROXY value")
|
||||||
|
assert testing_no_proxy, "TESTING_NO_PROXY envar missing"
|
||||||
|
|
||||||
|
# Send request and check the response
|
||||||
|
|
||||||
|
print("sending first request")
|
||||||
|
assert requests.get(server_url, timeout=2).text.startswith("Hello")
|
||||||
|
|
||||||
|
|
||||||
|
# Part two, dummy proxy causes timeout, that's fine
|
||||||
|
|
||||||
|
# Start proxy
|
||||||
|
print("starting proxy")
|
||||||
|
p = subprocess.Popen(["nc", "-k", "-l", "10000"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
|
subprocesses.append(p)
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
# Set proxy to environment
|
||||||
|
os.environ["HTTP_PROXY"] = "http://127.0.0.1:10000"
|
||||||
|
|
||||||
|
# Send request, expect timeout because proxy is dummy and do not respond
|
||||||
|
print("Sending request via proxy")
|
||||||
|
try:
|
||||||
|
requests.get(server_url, timeout=2)
|
||||||
|
except requests.exceptions.ReadTimeout:
|
||||||
|
print("timeout is fine, expected")
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Part three, NO_PROXY bypass the proxy and the request should work as before
|
||||||
|
|
||||||
|
os.environ["NO_PROXY"] = testing_no_proxy or no_proxy
|
||||||
|
|
||||||
|
print("sending last request")
|
||||||
|
try:
|
||||||
|
assert requests.get(server_url, timeout=2).text.startswith("Hello")
|
||||||
|
except requests.exceptions.ReadTimeout:
|
||||||
|
print("PROBLEM - the last request timed out, NO_PROXY setting does not work!")
|
||||||
|
raise SystemExit(1)
|
||||||
|
else:
|
||||||
|
print("OK - NO_PROXY setting works fine - the last request bypassed proxy")
|
25
tests/ipv6-noproxy-cidr/server.py
Normal file
25
tests/ipv6-noproxy-cidr/server.py
Normal file
@ -0,0 +1,25 @@
|
|||||||
|
from http.server import SimpleHTTPRequestHandler, HTTPServer
|
||||||
|
import socket
|
||||||
|
import sys
|
||||||
|
class IPv6OnlyHandler(SimpleHTTPRequestHandler):
|
||||||
|
def do_GET(self):
|
||||||
|
client_address = self.client_address[0]
|
||||||
|
# Check if the client address is an IPv6 address
|
||||||
|
if ":" in client_address:
|
||||||
|
self.send_response(200)
|
||||||
|
self.send_header('Content-type', 'text/plain')
|
||||||
|
self.end_headers()
|
||||||
|
self.wfile.write(b'Hello')
|
||||||
|
else:
|
||||||
|
# Raise a RuntimeError if the client address is an IPv4 address
|
||||||
|
raise SystemExit(f"IPv4 address ({client_address}) detected. Only IPv6 addresses are allowed.")
|
||||||
|
|
||||||
|
class HTTPServerV6(HTTPServer):
|
||||||
|
address_family = socket.AF_INET6
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
# Use IPv6 address and port 8080
|
||||||
|
server_address = ('::', 8888)
|
||||||
|
httpd = HTTPServerV6(server_address, IPv6OnlyHandler)
|
||||||
|
print('Server running on port 8888...')
|
||||||
|
httpd.serve_forever()
|
Loading…
Reference in New Issue
Block a user