emdee/exclude_badExits
1
0
Fork 0
Code Issues Pull requests Releases Packages Wiki Activity

Download

Browse source
This commit is contained in:
emdee 2022-11-08 14:15:05 +00:00
parent 2be5e6e66e
commit 0198994486
3 changed files with 588 additions and 233 deletions
Download patch file Download diff file Expand all files Collapse all files

156
trustor_poc.py
View file

@ -1,22 +1,26 @@
# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -# -*- coding: utf-8 -*-
# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -
import os
import sys
import datetime
import requests
from stem.control import Controller
from stem.util.tor_tools import *
from urllib.parse import urlparse
import requests
import datetime
try:
# unbound is not on pypi
from unbound import ub_ctx,RR_TYPE_TXT,RR_CLASS_IN
except:
ub_ctx = RR_TYPE_TXT = RR_CLASS_IN = None
global LOG
import logging
import warnings
warnings.filterwarnings('ignore')
LOG = logging.getLogger()
# download this python library from
# https://github.com/erans/torcontactinfoparser
#sys.path.append('/home/....')
@ -24,15 +28,6 @@ try:
from torcontactinfo import TorContactInfoParser
except:
TorContactInfoParser = None
# tor ControlPort IP
controller_address = '127.0.0.1'
dnssec_DS_file = 'dnssec-root-trust'
# this is not the system wide /etc/resolv.conf
# use dnscrypt-proxy to encrypt your DNS and route it via tor's SOCKSPort
libunbound_resolv_file = 'resolv.conf'
# for now we support max_depth = 0 only
# this PoC version has no support for recursion
@ -42,9 +37,6 @@ supported_max_depths = ['0']
# https://github.com/nusenu/ContactInfo-Information-Sharing-Specification#ciissversion
accepted_ciissversions = ['2']
# https://github.com/nusenu/ContactInfo-Information-Sharing-Specification#proof
accepted_proof_types = ['uri-rsa','dns-rsa']
# https://stackoverflow.com/questions/2532053/validate-a-hostname-string
# FIXME this check allows non-fqdn names
def is_valid_hostname(hostname):
@ -56,7 +48,7 @@ def is_valid_hostname(hostname):
return all(allowed.match(x) for x in hostname.split("."))
def read_local_trust_config(trust_config='trust_config'):
def read_local_trust_config(trust_config):
'''
reads a local configuration file containing trusted domains
and returns them in an array
@ -103,28 +95,28 @@ def read_local_validation_cache(validation_cache_file, trusted_domains=[]):
result = []
if trusted_domains == []:
return result
if (os.path.isfile(validation_cache_file)):
f = open(validation_cache_file)
for line in f:
line = line.strip()
if line[0] == '#':
continue
try:
domain, fingerprint, prooftype, dt = line.split(':')
except:
LOG.error('invalid trust cache entry detected: %s aborting!' % line)
sys.exit(12)
if os.path.isfile(validation_cache_file):
with open(validation_cache_file, 'rt') as f:
for line in f:
line = line.strip()
if line[0] == '#':
continue
try:
domain, fingerprint, prooftype, dt = line.split(':')
except:
LOG.error('invalid trust cache entry detected: %s aborting!' % line)
sys.exit(12)
if domain in trusted_domains:
result.append(fingerprint)
else:
print('ignoring cached entry for untrusted domain %s' % domain)
if domain in trusted_domains:
result.append(fingerprint)
else:
LOG.warn('ignoring cached entry for untrusted domain %s' % domain)
else:
print("Validation cache file not present. It will be created.")
LOG.info("Validation cache file not present. It will be created.")
return result
def get_controller(address='127.0.0.1',port=9151,password=''):
def get_controller(address='127.0.0.1', port=9151, password=''):
'''
connects to a local tor client via the tor ControlPort
and returns a controller that allows us to easily set specific tor
@ -136,7 +128,7 @@ def get_controller(address='127.0.0.1',port=9151,password=''):
controller = Controller.from_port(address=address, port=port)
controller.authenticate(password=password)
except Exception as e:
LOG.error(f'Failed to connect to the tor process, {e}')
LOG.error(f"Failed to connect to the tor process, {e}")
sys.exit(1)
if not controller.is_set('UseMicrodescriptors'):
@ -155,6 +147,9 @@ def find_validation_candidates(controller, trusted_domains=[],validation_cache=[
example content:
{ 'emeraldonion.org' : { 'uri-rsa': ['044600FD968728A6F220D5347AD897F421B757C0', '09DCA3360179C6C8A5A20DDDE1C54662965EF1BA']}}
'''
# https://github.com/nusenu/ContactInfo-Information-Sharing-Specification#proof
accepted_proof_types = ['uri-rsa','dns-rsa']
result = {}
@ -207,7 +202,7 @@ def find_validation_candidates(controller, trusted_domains=[],validation_cache=[
result[domain] = {prooftype : [fingerprint]}
return result
def lDownloadUrlFps(domain, timeout=20, host='127.0.0.1', port=9050):
def lDownloadUrlFps(domain, sCAfile, timeout=30, host='127.0.0.1', port=9050):
uri="https://"+domain+"/.well-known/tor-relay/rsa-fingerprint.txt"
# socks proxy used for outbound web requests (for validation of proofs)
proxy = {'https': 'socks5h://' +host +':' +str(port)}
@ -217,33 +212,53 @@ def lDownloadUrlFps(domain, timeout=20, host='127.0.0.1', port=9050):
LOG.debug("fetching %s...." % uri)
try:
# grr. fix urllib3
# urllib3.connection WARNING Certificate did not match expected hostname:
head = requests.head(uri, timeout=timeout, proxies=proxy, headers=headers)
except Exception as e:
print("HTTP HEAD request failed for %s" % uri)
print(e)
LOG.warn(f"HTTP HEAD request failed for {uri} {e}")
head = None
return []
if head.status_code != 200:
return []
if not head.headers['Content-Type'].startswith('text/plain'):
return []
assert os.path.exists(sCAfile), sCAfile
try:
fullfile = requests.get(uri, proxies=proxy, timeout=10, headers=headers)
from https_adapter import HTTPSAdapter
except Exception as e:
LOG.warn(f"Could not import HTTPSAdapter {e}")
HTTPSAdapter = None
HTTPSAdapter = None
try:
with requests.sessions.Session() as session:
if HTTPSAdapter:
# FixMe: upgrade to TLS1.3
session.mount("https://", HTTPSAdapter(pool_maxsize=1,
max_retries=3,))
fullfile = session.request(method="get", url=uri,
proxies=proxy, timeout=timeout,
headers=headers,
allow_redirects=False,
verify=True
)
except:
print("HTTP GET request failed for %s" % uri)
LOG.warn("HTTP GET request failed for %s" % uri)
return []
if fullfile.status_code != 200 or not fullfile.headers['Content-Type'].startswith('text/plain'):
return []
#check for redirects (not allowed as per spec)
if fullfile.url != uri:
LOG.error('Redirect detected %s vs %s (final)' % (uri, fullfile.url))
return []
well_known_content = [i.strip() for i in fullfile.text.upper().split('\n')]
well_known_content = fullfile.text.upper().strip().split('\n')
well_known_content = [i for i in well_known_content if i and len(i) == 40]
return well_known_content
def validate_proofs(candidates, validation_cache_file):
def validate_proofs(candidates, validation_cache_file, timeout=20, host='127.0.0.1', port=9050):
'''
This function takes the return value of find_validation_candidates()
and validated them according to their proof type (uri-rsa, dns-rsa)
@ -257,7 +272,7 @@ def validate_proofs(candidates, validation_cache_file):
for domain in candidates.keys():
for prooftype in candidates[domain].keys():
if prooftype == 'uri-rsa':
well_known_content = lDownloadUrlFps(domain, timeout=20, host='127.0.0.1', port=9050)
well_known_content = lDownloadUrlFps(domain, timeout=timeout, host=host, port=port)
for fingerprint in candidates[domain][prooftype]:
if fingerprint in well_known_content:
# write cache entry
@ -268,7 +283,10 @@ def validate_proofs(candidates, validation_cache_file):
elif prooftype == 'dns-rsa' and ub_ctx:
for fingerprint in candidates[domain][prooftype]:
fp_domain = fingerprint+'.'+domain
if dns_validate(fp_domain):
if idns_validate(fp_domain,
libunbound_resolv_file='resolv.conf',
dnssec_DS_file='dnssec-root-trust',
) == 0:
count += 1
f.write('%s:%s:%s:%s\n' % (domain, fingerprint, prooftype, dt_utc))
else:
@ -276,7 +294,10 @@ def validate_proofs(candidates, validation_cache_file):
f.close()
LOG.info('successfully validated %s new (not yet validated before) relays' % count)
def dns_validate(domain):
def idns_validate(domain,
libunbound_resolv_file='resolv.conf',
dnssec_DS_file='dnssec-root-trust',
):
'''
performs DNS TXT lookups and verifies the reply
- is DNSSEC valid and
@ -284,27 +305,31 @@ def dns_validate(domain):
- the DNS record contains a hardcoded string as per specification
https://nusenu.github.io/ContactInfo-Information-Sharing-Specification/#dns-rsa
'''
if not ub_ctx: return False
if not ub_ctx: return -1
# this is not the system wide /etc/resolv.conf
# use dnscrypt-proxy to encrypt your DNS and route it via tor's SOCKSPort
ctx = ub_ctx()
if (os.path.isfile(libunbound_resolv_file)):
ctx.resolvconf(libunbound_resolv_file)
else:
LOG.error('libunbound resolv config file: "%s" is missing, aborting!' % libunbound_resolv_file)
sys.exit(5)
return 5
if (os.path.isfile(dnssec_DS_file)):
ctx.add_ta_file(dnssec_DS_file)
else:
LOG.error('DNSSEC trust anchor file "%s" is missing, aborting!' % dnssec_DS_file)
sys.exit(6)
return 6
status, result = ctx.resolve(domain, RR_TYPE_TXT, RR_CLASS_IN)
if status == 0 and result.havedata:
if len(result.rawdata) == 1 and result.secure:
# ignore the first byte, it is the TXT length
if result.data.as_raw_data()[0][1:] == b'we-run-this-tor-relay':
return True
return False
return 0
return 1
def configure_tor(controller, trusted_fingerprints, exitonly=True):
'''
@ -317,32 +342,41 @@ def configure_tor(controller, trusted_fingerprints, exitonly=True):
relay_count = len(trusted_fingerprints)
if relay_count < 41:
print('Too few trusted relays (%s), aborting!' % relay_count)
LOG.error('Too few trusted relays (%s), aborting!' % relay_count)
sys.exit(15)
try:
controller.set_conf('ExitNodes', trusted_fingerprints)
print('limited exits to %s relays' % relay_count)
LOG.error('limited exits to %s relays' % relay_count)
except Exception as e:
print('Failed to set ExitNodes tor config to trusted relays')
print(e)
LOG.exception('Failed to set ExitNodes tor config to trusted relays')
sys.exit(20)
if __name__ == '__main__':
trust_config = 'trust_config'
assert os.path.exists(trust_config)
trusted_domains = read_local_trust_config(trust_config)
validation_cache_file = 'validation_cache'
trusted_fingerprints = read_local_validation_cache(validation_cache_file,
trusted_domains=trusted_domains)
# tor ControlPort password
controller_password=''
# tor ControlPort IP
controller_address = '127.0.0.1'
timeout = 20
port = 9050
controller = get_controller(address=controller_address,password=controller_password)
r = find_validation_candidates(controller,validation_cache=trusted_fingerprints,trusted_domains=trusted_domains)
validate_proofs(r, validation_cache_file)
r = find_validation_candidates(controller,
validation_cache=trusted_fingerprints,
trusted_domains=trusted_domains)
validate_proofs(r, validation_cache_file,
timeout=timeout,
host=controller_address,
port=port)
# refresh list with newly validated fingerprints
trusted_fingerprints = read_local_validation_cache(trusted_domains=trusted_domains)
trusted_fingerprints = read_local_validation_cache(validation_cache_file,
trusted_domains=trusted_domains)
configure_tor(controller, trusted_fingerprints)