https-proxy: use IP address and cert with ip in alt names

- improve info logging when peer verification fails to indicate
  if DNS name or ip address has been tried to match
- add test case for contacting https proxy with ip address
- add pytest env check on loaded credentials and re-issue
  when they are no longer valid
- disable proxy ip address test for bearssl, since not supported there

Ref: #12831
Closes #12838
This commit is contained in:
Stefan Eissing 2024-02-01 10:51:45 +01:00 committed by Daniel Stenberg
parent 4f79455877
commit c177e1944c
No known key found for this signature in database
GPG key ID: 5CC908FDB71E12C2
5 changed files with 50 additions and 14 deletions

View file

@ -2242,9 +2242,11 @@ CURLcode Curl_ossl_verifyhost(struct Curl_easy *data, struct connectdata *conn,
/* an alternative name matched */ /* an alternative name matched */
; ;
else if(dNSName || iPAddress) { else if(dNSName || iPAddress) {
infof(data, " subjectAltName does not match %s", peer->dispname); infof(data, " subjectAltName does not match %s %s",
peer->is_ip_address? "ip address" : "host name", peer->dispname);
failf(data, "SSL: no alternative certificate subject name matches " failf(data, "SSL: no alternative certificate subject name matches "
"target host name '%s'", peer->dispname); "target %s '%s'",
peer->is_ip_address? "ip address" : "host name", peer->dispname);
result = CURLE_PEER_FAILED_VERIFICATION; result = CURLE_PEER_FAILED_VERIFICATION;
} }
else { else {

View file

@ -70,8 +70,7 @@ class TestProxy:
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support') reason='curl lacks HTTPS-proxy support')
@pytest.mark.parametrize("proto", ['http/1.1', 'h2']) @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") def test_10_02_proxys_down(self, env: Env, httpd, proto, repeat):
def test_10_02_proxys_down(self, env: Env, httpd, nghttpx_fwd, proto, repeat):
if proto == 'h2' and not env.curl_uses_lib('nghttp2'): if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2') pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env) curl = CurlClient(env=env)
@ -349,3 +348,20 @@ class TestProxy:
extra_args=x2_args) extra_args=x2_args)
r2.check_response(count=2, http_status=200) r2.check_response(count=2, http_status=200)
assert r2.total_connects == 2 assert r2.total_connects == 2
# download via https: proxy (no tunnel) using IP address
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=Env.curl_uses_lib('bearssl'), reason="ip address cert verification not supported")
@pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
def test_10_14_proxys_ip_addr(self, env: Env, httpd, proto, repeat):
if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proto=proto, use_ip=True)
r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
extra_args=xargs)
r.check_response(count=1, http_status=200,
protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')

View file

@ -24,6 +24,7 @@
# #
########################################################################### ###########################################################################
# #
import ipaddress
import os import os
import re import re
from datetime import timedelta, datetime from datetime import timedelta, datetime
@ -79,6 +80,7 @@ class CertificateSpec:
valid_from: timedelta = timedelta(days=-1), valid_from: timedelta = timedelta(days=-1),
valid_to: timedelta = timedelta(days=89), valid_to: timedelta = timedelta(days=89),
client: bool = False, client: bool = False,
check_valid: bool = True,
sub_specs: Optional[List['CertificateSpec']] = None): sub_specs: Optional[List['CertificateSpec']] = None):
self._name = name self._name = name
self.domains = domains self.domains = domains
@ -89,6 +91,7 @@ class CertificateSpec:
self.valid_from = valid_from self.valid_from = valid_from
self.valid_to = valid_to self.valid_to = valid_to
self.sub_specs = sub_specs self.sub_specs = sub_specs
self.check_valid = check_valid
@property @property
def name(self) -> Optional[str]: def name(self) -> Optional[str]:
@ -202,7 +205,8 @@ class Credentials:
creds = None creds = None
if self._store: if self._store:
creds = self._store.load_credentials( creds = self._store.load_credentials(
name=spec.name, key_type=key_type, single_file=spec.single_file, issuer=self) name=spec.name, key_type=key_type, single_file=spec.single_file,
issuer=self, check_valid=spec.check_valid)
if creds is None: if creds is None:
creds = TestCA.create_credentials(spec=spec, issuer=self, key_type=key_type, creds = TestCA.create_credentials(spec=spec, issuer=self, key_type=key_type,
valid_from=spec.valid_from, valid_to=spec.valid_to) valid_from=spec.valid_from, valid_to=spec.valid_to)
@ -303,13 +307,18 @@ class CertStore:
def load_credentials(self, name: str, key_type=None, def load_credentials(self, name: str, key_type=None,
single_file: bool = False, single_file: bool = False,
issuer: Optional[Credentials] = None): issuer: Optional[Credentials] = None,
check_valid: bool = False):
cert_file = self.get_cert_file(name=name, key_type=key_type) cert_file = self.get_cert_file(name=name, key_type=key_type)
pkey_file = cert_file if single_file else self.get_pkey_file(name=name, key_type=key_type) pkey_file = cert_file if single_file else self.get_pkey_file(name=name, key_type=key_type)
comb_file = self.get_combined_file(name=name, key_type=key_type) comb_file = self.get_combined_file(name=name, key_type=key_type)
if os.path.isfile(cert_file) and os.path.isfile(pkey_file): if os.path.isfile(cert_file) and os.path.isfile(pkey_file):
cert = self.load_pem_cert(cert_file) cert = self.load_pem_cert(cert_file)
pkey = self.load_pem_pkey(pkey_file) pkey = self.load_pem_pkey(pkey_file)
if check_valid and \
((cert.not_valid_after < datetime.now()) or
(cert.not_valid_before > datetime.now())):
return None
creds = Credentials(name=name, cert=cert, pkey=pkey, issuer=issuer) creds = Credentials(name=name, cert=cert, pkey=pkey, issuer=issuer)
creds.set_store(self) creds.set_store(self)
creds.set_files(cert_file, pkey_file, comb_file) creds.set_files(cert_file, pkey_file, comb_file)
@ -426,6 +435,13 @@ class TestCA:
@staticmethod @staticmethod
def _add_leaf_usages(csr: Any, domains: List[str], issuer: Credentials) -> Any: def _add_leaf_usages(csr: Any, domains: List[str], issuer: Credentials) -> Any:
names = []
for name in domains:
try:
names.append(x509.IPAddress(ipaddress.ip_address(name)))
except:
names.append(x509.DNSName(name))
return csr.add_extension( return csr.add_extension(
x509.BasicConstraints(ca=False, path_length=None), x509.BasicConstraints(ca=False, path_length=None),
critical=True, critical=True,
@ -435,8 +451,7 @@ class TestCA:
x509.SubjectKeyIdentifier).value), x509.SubjectKeyIdentifier).value),
critical=False critical=False
).add_extension( ).add_extension(
x509.SubjectAlternativeName([x509.DNSName(domain) for domain in domains]), x509.SubjectAlternativeName(names), critical=True,
critical=True,
).add_extension( ).add_extension(
x509.ExtendedKeyUsage([ x509.ExtendedKeyUsage([
ExtendedKeyUsageOID.SERVER_AUTH, ExtendedKeyUsageOID.SERVER_AUTH,

View file

@ -393,20 +393,22 @@ class CurlClient:
return os.makedirs(path) return os.makedirs(path)
def get_proxy_args(self, proto: str = 'http/1.1', def get_proxy_args(self, proto: str = 'http/1.1',
proxys: bool = True, tunnel: bool = False): proxys: bool = True, tunnel: bool = False,
use_ip: bool = False):
proxy_name = '127.0.0.1' if use_ip else self.env.proxy_domain
if proxys: if proxys:
pport = self.env.pts_port(proto) if tunnel else self.env.proxys_port pport = self.env.pts_port(proto) if tunnel else self.env.proxys_port
xargs = [ xargs = [
'--proxy', f'https://{self.env.proxy_domain}:{pport}/', '--proxy', f'https://{proxy_name}:{pport}/',
'--resolve', f'{self.env.proxy_domain}:{pport}:127.0.0.1', '--resolve', f'{proxy_name}:{pport}:127.0.0.1',
'--proxy-cacert', self.env.ca.cert_file, '--proxy-cacert', self.env.ca.cert_file,
] ]
if proto == 'h2': if proto == 'h2':
xargs.append('--proxy-http2') xargs.append('--proxy-http2')
else: else:
xargs = [ xargs = [
'--proxy', f'http://{self.env.proxy_domain}:{self.env.proxy_port}/', '--proxy', f'http://{proxy_name}:{self.env.proxy_port}/',
'--resolve', f'{self.env.proxy_domain}:{self.env.proxy_port}:127.0.0.1', '--resolve', f'{proxy_name}:{self.env.proxy_port}:127.0.0.1',
] ]
if tunnel: if tunnel:
xargs.append('--proxytunnel') xargs.append('--proxytunnel')

View file

@ -31,6 +31,7 @@ import socket
import subprocess import subprocess
import sys import sys
from configparser import ConfigParser, ExtendedInterpolation from configparser import ConfigParser, ExtendedInterpolation
from datetime import timedelta
from typing import Optional from typing import Optional
import pytest import pytest
@ -133,7 +134,7 @@ class EnvConfig:
self.cert_specs = [ self.cert_specs = [
CertificateSpec(domains=[self.domain1, 'localhost'], key_type='rsa2048'), CertificateSpec(domains=[self.domain1, 'localhost'], key_type='rsa2048'),
CertificateSpec(domains=[self.domain2], key_type='rsa2048'), CertificateSpec(domains=[self.domain2], key_type='rsa2048'),
CertificateSpec(domains=[self.proxy_domain], key_type='rsa2048'), CertificateSpec(domains=[self.proxy_domain, '127.0.0.1'], key_type='rsa2048'),
CertificateSpec(name="clientsX", sub_specs=[ CertificateSpec(name="clientsX", sub_specs=[
CertificateSpec(name="user1", client=True), CertificateSpec(name="user1", client=True),
]), ]),