feat: Add send page
This commit is contained in:
136
domainLookup.py
Normal file
136
domainLookup.py
Normal file
@@ -0,0 +1,136 @@
|
||||
import dns.resolver
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
import tempfile
|
||||
import subprocess
|
||||
import binascii
|
||||
import datetime
|
||||
import dns.asyncresolver
|
||||
import httpx
|
||||
from requests_doh import DNSOverHTTPSSession, add_dns_provider, remove_dns_provider
|
||||
|
||||
|
||||
def hip2(domain: str):
|
||||
domain_check = False
|
||||
try:
|
||||
# Get the IP
|
||||
ip = resolve_with_doh(domain)
|
||||
|
||||
# Run the openssl s_client command
|
||||
s_client_command = ["openssl","s_client","-showcerts","-connect",f"{ip}:443","-servername",domain,]
|
||||
|
||||
s_client_process = subprocess.Popen(s_client_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
|
||||
s_client_output, _ = s_client_process.communicate(input=b"\n")
|
||||
|
||||
certificates = []
|
||||
current_cert = ""
|
||||
for line in s_client_output.split(b"\n"):
|
||||
current_cert += line.decode("utf-8") + "\n"
|
||||
if "-----END CERTIFICATE-----" in line.decode("utf-8"):
|
||||
certificates.append(current_cert)
|
||||
current_cert = ""
|
||||
|
||||
# Remove anything before -----BEGIN CERTIFICATE-----
|
||||
certificates = [cert[cert.find("-----BEGIN CERTIFICATE-----"):] for cert in certificates]
|
||||
|
||||
if certificates:
|
||||
cert = certificates[0]
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_cert_file:
|
||||
temp_cert_file.write(cert)
|
||||
temp_cert_file.seek(0) # Move back to the beginning of the temporary file
|
||||
|
||||
tlsa_command = ["openssl","x509","-in",temp_cert_file.name,"-pubkey","-noout","|","openssl","pkey","-pubin","-outform","der","|","openssl","dgst","-sha256","-binary",]
|
||||
|
||||
tlsa_process = subprocess.Popen(" ".join(tlsa_command), shell=True, stdout=subprocess.PIPE)
|
||||
tlsa_output, _ = tlsa_process.communicate()
|
||||
|
||||
tlsa_server = "3 1 1 " + binascii.hexlify(tlsa_output).decode("utf-8")
|
||||
|
||||
|
||||
# Get domains
|
||||
cert_obj = x509.load_pem_x509_certificate(cert.encode("utf-8"), default_backend())
|
||||
|
||||
domains = []
|
||||
for ext in cert_obj.extensions:
|
||||
if ext.oid == x509.ExtensionOID.SUBJECT_ALTERNATIVE_NAME:
|
||||
san_list = ext.value.get_values_for_type(x509.DNSName)
|
||||
domains.extend(san_list)
|
||||
|
||||
# Extract the common name (CN) from the subject
|
||||
common_name = cert_obj.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
|
||||
if common_name:
|
||||
if common_name[0].value not in domains:
|
||||
domains.append(common_name[0].value)
|
||||
|
||||
if domains:
|
||||
if domain in domains:
|
||||
domain_check = True
|
||||
else:
|
||||
# Check if matching wildcard domain exists
|
||||
for d in domains:
|
||||
if d.startswith("*"):
|
||||
if domain.split(".")[1:] == d.split(".")[1:]:
|
||||
domain_check = True
|
||||
break
|
||||
|
||||
|
||||
expiry_date = cert_obj.not_valid_after
|
||||
# Check if expiry date is past
|
||||
if expiry_date < datetime.datetime.now():
|
||||
return "Hip2: Certificate is expired"
|
||||
|
||||
|
||||
else:
|
||||
return "Hip2: No certificate found"
|
||||
|
||||
try:
|
||||
# Check for TLSA record
|
||||
tlsa = resolve_TLSA_with_doh(domain)
|
||||
|
||||
|
||||
if not tlsa:
|
||||
return "Hip2: TLSA lookup failed"
|
||||
else:
|
||||
if tlsa_server == str(tlsa):
|
||||
if domain_check:
|
||||
# Get the Hip2 addresss from /.well-known/wallets/HNS
|
||||
add_dns_provider("HNSDoH", "https://hnsdoh.com/dns-query")
|
||||
|
||||
session = DNSOverHTTPSSession("HNSDoH")
|
||||
r = session.get(f"https://{domain}/.well-known/wallets/HNS",verify=False)
|
||||
return r.text
|
||||
else:
|
||||
return "Hip2: TLSA record matches certificate, but domain does not match certificate"
|
||||
|
||||
else:
|
||||
return "Hip2: TLSA record does not match certificate"
|
||||
|
||||
except Exception as e:
|
||||
return "Hip2: TLSA lookup failed with error: " + str(e)
|
||||
|
||||
|
||||
|
||||
|
||||
# Catch all exceptions
|
||||
except Exception as e:
|
||||
return "Hip2: " + str(e)
|
||||
|
||||
|
||||
def resolve_with_doh(query_name, doh_url="https://hnsdoh.com/dns-query"):
|
||||
with httpx.Client() as client:
|
||||
q = dns.message.make_query(query_name, dns.rdatatype.A)
|
||||
r = dns.query.https(q, doh_url, session=client)
|
||||
|
||||
ip = r.answer[0][0].address
|
||||
return ip
|
||||
|
||||
def resolve_TLSA_with_doh(query_name, doh_url="https://hnsdoh.com/dns-query"):
|
||||
query_name = "_443._tcp." + query_name
|
||||
with httpx.Client() as client:
|
||||
q = dns.message.make_query(query_name, dns.rdatatype.TLSA)
|
||||
r = dns.query.https(q, doh_url, session=client)
|
||||
|
||||
tlsa = r.answer[0][0]
|
||||
return tlsa
|
||||
|
||||
Reference in New Issue
Block a user