Commit 1693dd9d authored by Łukasz Nowak's avatar Łukasz Nowak

caddy-frontend/test: Replace forcediphttpsadapter

forcediphttpsadapter does not work with requests' SourceAddressAdapter,
so by following https://stackoverflow.com/a/44378047/9256748 implement
own curl's --resolve style.
parent 7fae24d5
......@@ -46,7 +46,6 @@ setup(name=name,
# ipaddress is patching IPAddress so IPv6 in SSL certificates
# match works
'ipaddress >= 1.0.22',
'forcediphttpsadapter',
'requests-toolbelt',
'supervisor',
# caucase needed to connect to the KeDiFa caucase
......
......@@ -37,7 +37,6 @@ from unittest import skip
import ssl
from BaseHTTPServer import HTTPServer
from BaseHTTPServer import BaseHTTPRequestHandler
from forcediphttpsadapter.adapters import ForcedIPHTTPSAdapter
import time
import tempfile
import ipaddress
......@@ -48,6 +47,7 @@ import re
from slapos.recipe.librecipe import generateHashFromFiles
import xml.etree.ElementTree as ET
import urlparse
import socket
try:
......@@ -81,6 +81,21 @@ KEDIFA_PORT = '15080'
# has to be not partition one
SOURCE_IP = '127.0.0.1'
# "--resolve" inspired from https://stackoverflow.com/a/44378047/9256748
DNS_CACHE = {}
def add_custom_dns(domain, port, ip):
port = int(port)
key = (domain, port)
value = (socket.AF_INET, 1, 6, '', (ip, port))
DNS_CACHE[key] = [value]
def new_getaddrinfo(*args):
return DNS_CACHE[args[:2]]
# for development: debugging logs and install Ctrl+C handler
if os.environ.get('SLAPOS_TEST_DEBUG'):
import logging
......@@ -417,21 +432,23 @@ def fakeHTTPSResult(domain, real_ip, path, port=HTTPS_PORT,
headers.setdefault('X-Forwarded-Port', '17')
session = requests.Session()
session.mount(
'https://%s:%s' % (domain, port),
ForcedIPHTTPSAdapter(
dest_ip=real_ip))
if source_ip is not None:
new_source = source.SourceAddressAdapter(source_ip)
session.mount('http://', new_source)
session.mount('https://', new_source)
return session.get(
'https://%s:%s/%s' % (domain, port, path),
verify=False,
allow_redirects=False,
headers=headers,
cookies=cookies
)
socket_getaddrinfo = socket.getaddrinfo
try:
add_custom_dns(domain, port, real_ip)
socket.getaddrinfo = new_getaddrinfo
return session.get(
'https://%s:%s/%s' % (domain, port, path),
verify=False,
allow_redirects=False,
headers=headers,
cookies=cookies
)
finally:
socket.getaddrinfo = socket_getaddrinfo
def fakeHTTPResult(domain, real_ip, path, port=HTTP_PORT,
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment