tuxmain/tlsbench
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
tlsbench/exp_proxies.py
Pascal Engélibert a71491c068 Initial commit
2025-11-05 14:28:26 +01:00

535 lines
17 KiB
Python
Raw Blame History

#!/usr/bin/python3
# OLD VERSION
# using 3 proxies between actual browser and actual server
import os, sys, subprocess
REPODIR = "/home/tuxmain/Documents/doc/tlsbench"
P2_SSH = "exp@p2"
P2_PSW = "exp"
P2_REPODIR = "/home/exp/exp"
EXPDIR = "/dev/shm/exp"
DOMAINS_ = [
# Apple
"apple.com", "www.apple.com", "graffiti-tags.apple.com",
"store.storeimages.cdn-apple.com",
"mzstatic.com", "is1-ssl.mzstatic.com"
# Youtube video
"youtube.com", "www.youtube.com", "i.ytimg.com", "fonts.gstatic.com", "www.google.com", "accounts.google.com", "yt3.ggpht.com", "www.gstatic.com", "fonts.googleapis.com", "www.googleadservices.com",
# W3C
"w3.org", "www.w3.org",
# Amazon
"amazon.com", "www.amazon.com",
# JA3 fingerprint test
"www.edgecomputing.live",
# JA4 fingerprint test
"ja4db.com",
# Local test
"foo.home",
# Peertube
"flim.txmn.tk",
]
#URL = "https://www.youtube.com/watch?v=IBWezPN4Ep8"
#URL = "https://www.edgecompute.live/tls/ja3"
#URL = "https://ja4db.com/id/ja4/"
#URL = "https://www.w3.org/"
#URL = "http://foo.home:8080/?hello=hi"
#URL = "https://www.apple.com/"
#URL = "https://www.amazon.com/"
URL = "https://flim.txmn.tk/w/4zpfvGB72oTL4hcSqdAwx8"
URLS = [
# Heavy showcase website
"https://www.apple.com/",
# Light showcase website
"https://librezo.fr/",
# Long Wikipedia article
"https://fr.wikipedia.org/wiki/Sp%C3%A9cial:Recherche?search=Victor+Hugo&sourceid=Mozilla-search",
# Youtube video
"https://www.youtube.com/watch?v=IBWezPN4Ep8",
# Peertube video
"https://videos.domainepublic.net/videos/watch/eaee7866-d209-4e5c-b7b0-443395b79c82",
# Google search
"https://www.google.com/search?q=where+do+birds+go+when+it+rains"
]
CERT_SIGN_ALGS = [
"prime256v1", # widely used
"secp384r1", # rarely used but supported by browsers because it's NIST standard
#"secp521r1", # not supported by browsers because NIST said it was not needed
"rsa2048", "rsa3072", "rsa4096", # widely used
]
IMPLS = [
#"aws_lc_rs", # Amazon's Rust crypto widely used in Rust stuff
#"boring", # Google's fork of OpenSSL used in Chrome and Android
#"openssl", # widely used
"ring", # used in most Rust stuff
#"symcrypt", # Microsoft's crypto
#"wolfcrypt" # used in embedded (won't build with rpxy for now)
]
# Symmetric ciphers
# They also allow to choose the TLS version.
CIPHERS = [
# TLS 1.3
"AES_256_GCM_SHA384",
"AES_128_GCM_SHA256",
"CHACHA20_POLY1305_SHA256",
# TLS 1.2
# ECDSA vs RSA refers to the certificate signature algorithm.
# DH is EC is aither case, using the group specified below.
"ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,ECDHE_RSA_WITH_AES_256_GCM_SHA384",
"ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,ECDHE_RSA_WITH_AES_128_GCM_SHA256",
"ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256",
]
KEXES = [
"X25519",
"SECP256R1",
"SECP384R1",
]
# Testing all combinations would be too much. Instead we isolate independent parts.
EXPERIMENTS = {
# Compare ciphers among implementations and TLS versions
"impl-cipher-ver": {
"impls": IMPLS,
"urls": [URL],
"ciphers": [
"AES_128_GCM_SHA256",
"AES_256_GCM_SHA384",
"CHACHA20_POLY1305_SHA256",
"ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,ECDHE_RSA_WITH_AES_128_GCM_SHA256",
"ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,ECDHE_RSA_WITH_AES_256_GCM_SHA384",
"ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256",
],
"kexes": ["X25519"],
"cert": ["prime256v1"],
},
# Compare signatures among implementations and TLS versions
"impl-cert-ver": {
"impls": IMPLS,
"urls": [URL],
"ciphers": [
"AES_128_GCM_SHA256",
"ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,ECDHE_RSA_WITH_AES_128_GCM_SHA256",
],
"kexes": ["X25519"],
"cert": ["prime256v1", "secp384r1", "rsa2048", "rsa3072", "rsa4096"],
},
# Compare key exchange groups among implementations and TLS versions
"impl-kex-ver": {
"impls": IMPLS,
"urls": [URL],
"ciphers": [
"AES_128_GCM_SHA256",
"ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,ECDHE_RSA_WITH_AES_128_GCM_SHA256",
],
"kexes": ["X25519", "SECP256R1", "SECP384R1"],
"cert": ["prime256v1"],
},
}
DOMAINS = []
for domain in DOMAINS_:
if not domain in DOMAINS:
DOMAINS.append(domain)
# JS to redirect the target domains to local (bypass DNS without altering system's config or webpages or packets)
SCRIPT_FIREFOX_HOSTS = """const gOverride = Cc["@mozilla.org/network/native-dns-override;1"].getService(Ci.nsINativeDNSResolverOverride);
gOverride.clearOverrides();
var names = """+str(DOMAINS)+""";
for(var i in names) {
gOverride.addIPOverride(names[i], "127.0.0.1");
}
"""
def getargv(arg:str, default:str="", n:int=1, args:list=sys.argv) -> str:
if arg in args and len(args) > args.index(arg)+n:
return args[args.index(arg)+n]
else:
return default
def sh(cmds):
if type(cmds) == list:
for cmd in cmds:
os.system(cmd)
elif type(cmds) == str:
os.system(cmds)
else:
raise TypeError
def make_sk(outpath, alg):
sh({
"ed25519": [
f"openssl genpkey -out {outpath}.sec1 -algorithm ed25519",
f"openssl pkcs8 -topk8 -nocrypt -in {outpath}.sec1 -out {outpath} -outform PEM"
],
"prime256v1": [
f"openssl ecparam -genkey -name prime256v1 -noout -out {outpath}.sec1",
f"openssl pkcs8 -topk8 -nocrypt -in {outpath}.sec1 -out {outpath} -outform PEM"
],
"rsa2048": [
f"openssl genrsa -out {outpath} 2048"
],
"rsa3072": [
f"openssl genrsa -out {outpath} 3072"
],
"rsa4096": [
f"openssl genrsa -out {outpath} 4096"
],
"secp384r1": [
f"openssl ecparam -genkey -name secp384r1 -noout -out {outpath}.sec1",
f"openssl pkcs8 -topk8 -nocrypt -in {outpath}.sec1 -out {outpath} -outform PEM"
],
}[alg])
def make_ca_cert(outpath, skpath):
sh(f"openssl req -x509 -new -key {skpath} -sha256 -days 730 -out {outpath} -subj '/CN=Foobar Root CA/C=AT/ST=Lyon/L=Lyon/O=Foobar'")
def make_cert(outpath, skpath, capath, caskpath, name, domains):
sh(f"openssl req -new -key {skpath} -sha256 -out /tmp/tmp.csr -subj '/CN={name}/C=AT/ST=Lyon/L=Lyon/O={name}'")
ext = open("/tmp/tmp.v3.ext", "w")
ext.write("""authorityKeyIdentifier=keyid,issuer
basicConstraints=CA:FALSE
keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment
subjectAltName = @alt_names
[alt_names]
""")
i = 1
for domain in domains:
ext.write(f"DNS.{i} = {domain}\n")
i += 1
ext.write(f"DNS.{i} = {domain}.localhost\n")
i += 1
ext.write(f"DNS.{i} = {domain}.p2\n")
i += 1
ext.write(f"DNS.{i} = {domain}.p3\n")
i += 1
ext.close()
sh(f"openssl x509 -req -in /tmp/tmp.csr -CA {capath} -CAkey {caskpath} -CAcreateserial -out {outpath} -days 365 -sha256 -extfile /tmp/tmp.v3.ext")
def get_domain_root(domain):
last_dot = domain.rfind(".")
penultimate_dot = domain.rfind(".", 0, last_dot)
return domain[penultimate_dot+1:]
# Issue secret keys, CA cert and signed certs for given domains
# All using the same algorithm
def make_certs(outdir, domains, alg):
if outdir[-1] != "/":
outdir += "/"
make_sk(outdir+"ca.key", alg)
make_ca_cert(outdir+"ca.crt", outdir+"ca.key")
# Only make certs for root domains, and include subdomains
roots = {}
for domain in domains:
root = domain
if domain.count(".") > 1:
root = get_domain_root(domain)
if root in roots:
roots[root].append(domain)
else:
roots[root] = [domain]
for root in roots:
make_sk(outdir+root+".key", alg)
make_cert(outdir+root+".crt", outdir+root+".key", outdir+"ca.crt", outdir+"ca.key", root, roots[root])
RPXY_CONFIGS = {
"p1_plain": {
"listen_http": 80,
"listen_https": 443,
"app": """[apps.{app}]
server_name = "{domain}"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p2:42002" }}]}}]
[apps.{app}_localhost]
server_name = "{domain}.localhost"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p2:42002" }}]}}]
"""
},
"p1_tls": {
"listen_http": 80,
"listen_https": 443,
"app": """[apps.{app}]
server_name = "{domain}"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p2:43002", tls = true }}], set_host = "{domain}.p2"}}]
[apps.{app}_localhost]
server_name = "{domain}.localhost"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p2:43002", tls = true }}], set_host = "{domain}.p2"}}]
"""
},
"p2_plain": {
"listen_http": 42002,
"listen_https": 43002,
"app": """[apps.{app}]
server_name = "{domain}"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p3:42003" }}], set_host = "{domain}.p3"}}]
[apps.{app}_localhost]
server_name = "{domain}.localhost"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p3:42003" }}], set_host = "{domain}.p3"}}]
[apps.{app}_p2]
server_name = "{domain}.p2"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p3:42003" }}], set_host = "{domain}.p3"}}]
"""
},
"p2_tls": {
"listen_http": 42002,
"listen_https": 43002,
"app": """[apps.{app}]
server_name = "{domain}"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p3:43003", tls = true }}], set_host = "{domain}.p3"}}]
[apps.{app}_localhost]
server_name = "{domain}.localhost"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p3:43003", tls = true }}], set_host = "{domain}.p3"}}]
[apps.{app}_p2]
server_name = "{domain}.p2"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}.p3:43003", tls = true }}], set_host = "{domain}.p3"}}]
"""
},
"p3": {
"listen_http": 42003,
"listen_https": 43003,
"app": """[apps.{app}]
server_name = "{domain}"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}", tls = true }}], set_host = "{domain}"}}]
[apps.{app}_localhost]
server_name = "{domain}.localhost"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}", tls = true }}], set_host = "{domain}"}}]
[apps.{app}_p3]
server_name = "{domain}.p3"
tls = {{ tls_cert_path = "{cert}", tls_cert_key_path = "{key}", https_redirection = false }}
reverse_proxy = [{{ upstream = [{{ location = "{domain}", tls = true }}], set_host = "{domain}"}}]
"""
},
}
SETUPS = {
#"none": {
# "rpxy_configs": ["p1_plain", "p2_plain", "p3"],
#},
#"client": {
# "rpxy_configs": ["p1_plain", "p2_tls", "p3"],
#},
"server": {
"rpxy_configs": ["p1_tls", "p2_plain", "p3"],
},
}
def make_rpxy_config(outdir, domains, cryptodir):
if outdir[-1] != "/":
outdir += "/"
if cryptodir[-1] != "/":
cryptodir += "/"
for config_name in RPXY_CONFIGS:
config = RPXY_CONFIGS[config_name]
f = open(outdir+config_name+".toml", "w")
f.write("listen_port = {}\nlisten_port_tls = {}\n".format(config["listen_http"], config["listen_https"]))
for domain in domains:
app = domain.replace(".", "_")
root = get_domain_root(domain)
f.write(config["app"].format(
app=app,
domain=domain,
cert=cryptodir+root+".crt",
key=cryptodir+root+".key",
))
f.close()
def make_everything(expdir, domains):
os.makedirs(expdir, exist_ok=True)
if expdir[-1] != "/":
expdir += "/"
for alg in CERT_SIGN_ALGS:
algdir = expdir+"certs/"+alg
os.makedirs(algdir, exist_ok=True)
make_certs(algdir, domains, alg)
# this will be a symbolic link to the chosen certs directory
cryptodir = expdir+"current_certs"
for setup in SETUPS:
setupdir = expdir+"setups/"+setup
os.makedirs(setupdir, exist_ok=True)
make_rpxy_config(setupdir, domains, cryptodir)
def choose_cert_alg(expdir, alg):
if expdir[-1] != "/":
expdir += "/"
try:
os.unlink(expdir+"current_certs")
except FileNotFoundError as e:
pass
# WHY is dst pointing to src?????
os.symlink(expdir+"certs/"+alg, expdir+"current_certs", True)
def choose_impl(expdir, p, impl):
if expdir[-1] != "/":
expdir += "/"
os.symlink(os.getcwd()+"/rpxy_rustls_"+impl, expdir+str(p)+"_rpxy", False)
def run_rpxy(expdir, repodir, setup, p, impl, ciphers=None, kexes=None):
if expdir[-1] != "/":
expdir += "/"
repodir = repodir.removesuffix("/")
env = {"RUST_LOG": "debug"}
if ciphers:
env["CIPHERS"] = ",".join(ciphers)
if kexes:
env["KEXES"] = ",".join(KEXES)
print("RUN", [repodir+"/rpxy_rustls_"+impl, "--config", expdir+"setups/"+setup+"/"+p+".toml"])
return subprocess.Popen([repodir+"/rpxy_rustls_"+impl, "--config", expdir+"setups/"+setup+"/"+p+".toml"], env=env)
# https://stackoverflow.com/questions/8775598/how-to-start-a-background-process-with-nohup-using-fabric
def runbg(ssh, cmd, vars={}):
strvars = ""
for var in vars:
strvars += f"export {var}="+vars[var]+" && "
return ssh.run(f"{strvars}dtach -n `mktemp -u /tmp/dtach.XXXX` {cmd}")
def run_exp(ssh, expdir, p2_path, exps):
from selenium import webdriver
import time
import invoke
logfile = open(expdir+"/log-"+str(int(time.time())), "w")
driver = webdriver.Firefox()
driver.get("about:config")
driver.execute_script(SCRIPT_FIREFOX_HOSTS)
time.sleep(1)
for expname in exps:
exp = exps[expname]
for impl in exp["impls"]:
sh(f"killall rpxy_rustls_{impl}")
try:
ssh.run(f"killall rpxy_rustls_{impl}")
except invoke.exceptions.UnexpectedExit as e:
pass
for impl in exp["impls"]:
for alg in exp["cert"]:
for kex in exp["kexes"]:
for cipher in exp["ciphers"]:
choose_cert_alg(expdir, alg)
ssh.run(f"python {p2_path}/exp.py cert {alg}")
for setup in SETUPS:
print(f"EXPERIMENT {expname}: {impl} {alg} {kex} {cipher} {setup}")
setupdir = expdir+"setups/"+setup
p1 = run_rpxy(expdir, REPODIR, setup, SETUPS[setup]["rpxy_configs"][0], impl)
p3 = run_rpxy(expdir, REPODIR, setup, SETUPS[setup]["rpxy_configs"][2], impl)
p2_rpxy_config = SETUPS[setup]["rpxy_configs"][1]
vars = {"CIPHERS": cipher, "KEXES": kex}
runbg(ssh, f"{p2_path}/rpxy_rustls_{impl} --config {expdir}/setups/{setup}/{p2_rpxy_config}.toml --log-dir /dev/shm", vars)
start = time.time()
for url in exp["urls"]:
driver.get(url)
#time.sleep(20)
input("???")
end = time.time()
logfile.write(f"{expname} {impl} {alg} {kex} {cipher} {setup} {start} {end}\n")
# DO the experiment
p1.kill()
p3.kill()
ssh.run(f"killall rpxy_rustls_{impl}")
ssh.run(f"killall dtach")
logfile.flush()
driver.quit()
def update_certs():
import platform
dist = platform.freedesktop_os_release()["ID"]
if dist == "debian":
for alg in CERT_SIGN_ALGS:
sh([
f"sudo cp {EXPDIR}/certs/{alg}/ca.crt /usr/local/share/ca-certificates/ca-{alg}.crt",
f"sudo chown root:root /usr/local/share/ca-certificates/ca-{alg}.crt"
])
sh("sudo update-ca-certificates")
elif dist == "arch":
for alg in CERT_SIGN_ALGS:
sh([
f"sudo cp {EXPDIR}/certs/{alg}/ca.crt /etc/ca-certificates/trust-source/anchors/ca-{alg}.crt",
f"sudo chown root:root /etc/ca-certificates/trust-source/anchors/ca-{alg}.crt"
])
sh("sudo update-ca-trust extract")
# copy local dir src to remote parent dir dst (src's copy will be a subdir of dst)
def upload_dir(ssh, src, dst):
src = src.removesuffix("/")
src_name = src.split("/")[-1]
os.chdir(f"{src}/..")
os.system(f"tar -czf {src}/../tmp.tar.gz {src_name}")
print(ssh.put(src+"/../tmp.tar.gz", dst))
print(ssh.run(f"cd {dst} && tar -xf tmp.tar.gz"))
def connect_ssh():
import fabric
ssh_passphrase = "--passphrase" in sys.argv
connect_kwargs = {}
if ssh_passphrase:
import getpass
connect_kwargs["passphrase"] = getpass.getpass("Enter passphrase for SSH key: ")
if P2_PSW != None:
connect_kwargs["password"] = P2_PSW
ssh = fabric.Connection(P2_SSH, connect_kwargs=connect_kwargs)
return ssh
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] in ["h", "help", "?", "-h", "-help", "--help", "/?"]:
print("""Options:
make Create everything
cert <alg> Select cert signature algorithm
send
update-certs
run Run experiment
Signature algorithms:
{sig_algs}
Implementations:
{impls}
Run options:
--passphrase Prompt SSH key decryption passphrase (when using pubkey login)
""".format(
sig_algs=" ".join(CERT_SIGN_ALGS),
impls=" ".join(IMPLS),
))
exit()
opt = sys.argv[1]
if opt == "make":
make_everything(EXPDIR, DOMAINS)
elif opt == "cert":
alg = sys.argv[2]
if not alg in CERT_SIGN_ALGS:
print("Error: alg must be in", CERT_SIGN_ALGS)
exit(1)
choose_cert_alg(EXPDIR, alg)
elif opt == "send":
ssh = connect_ssh()
upload_dir(ssh, EXPDIR, "/dev/shm")
elif opt == "update-certs":
update_certs()
elif opt == "run":
ssh = connect_ssh()
run_exp(ssh, EXPDIR, P2_REPODIR, EXPERIMENTS)
else:
print("Unknown command, use help for help")
exit(1)
Reference in a new issue View git blame Copy permalink