from collections import defaultdict from functools import cache import json from flask import ( Flask, make_response, redirect, request, jsonify, render_template, send_from_directory, send_file, ) import os import json import requests import dns.resolver import dns.message import dns.query import dns.name import dns.rdatatype import ssl import dnslib import dnslib.dns import socket from datetime import datetime from dateutil import relativedelta app = Flask(__name__) node_names = { "18.169.98.42": "Easy HNS", "172.233.46.92": "EZ Domains", "194.50.5.27": "Nathan.Woodburn/", "139.177.195.185": "HNSCanada", "172.105.120.203": "EZ Domains", } node_locations = { "18.169.98.42": "England", "172.233.46.92": "Netherlands", "194.50.5.27": "Australia", "139.177.195.185": "Canada", "172.105.120.203": "Singapore", } nodes = [] last_log = datetime.now() - relativedelta.relativedelta(years=1) log_dir = "/data" if not os.path.exists(log_dir): if not os.path.exists("./logs"): os.mkdir("./logs") log_dir = "./logs" print(f"Log directory: {log_dir}", flush=True) def find(name, path): for root, dirs, files in os.walk(path): if name in files: return os.path.join(root, name) # Assets routes @app.route("/assets/") def send_report(path): if path.endswith(".json"): return send_from_directory( "templates/assets", path, mimetype="application/json" ) if os.path.isfile("templates/assets/" + path): return send_from_directory("templates/assets", path) # Try looking in one of the directories filename: str = path.split("/")[-1] if ( filename.endswith(".png") or filename.endswith(".jpg") or filename.endswith(".jpeg") or filename.endswith(".svg") ): if os.path.isfile("templates/assets/img/" + filename): return send_from_directory("templates/assets/img", filename) if os.path.isfile("templates/assets/img/favicon/" + filename): return send_from_directory("templates/assets/img/favicon", filename) return render_template("404.html"), 404 # region Special routes @app.route("/favicon.png") def faviconPNG(): return send_from_directory("templates/assets/img", "favicon.png") @app.route("/.well-known/") def wellknown(path): # Try to proxy to https://nathan.woodburn.au/.well-known/ req = requests.get(f"https://nathan.woodburn.au/.well-known/{path}") return make_response( req.content, 200, {"Content-Type": req.headers["Content-Type"]} ) # endregion # region Helper functions def get_node_list() -> list: ips = [] # Do a DNS lookup result: dns.resolver.Answer = dns.resolver.resolve("hnsdoh.com", "A") # Print the IP addresses for ipval in result: ips.append(ipval.to_text()) return ips def check_plain_dns(ip: str) -> bool: resolver = dns.resolver.Resolver() resolver.nameservers = [ip] try: result = resolver.resolve("1.wdbrn", "TXT") for txt in result: if "Test 1" in txt.to_text(): return True return False except Exception as e: print(e) return False def build_dns_query(domain: str, qtype: str = 'A'): """ Constructs a DNS query in binary wire format using dnslib. """ q = dnslib.DNSRecord.question(domain, qtype) return q.pack() def check_doh(ip: str) -> bool: status = False dns_query = build_dns_query("2.wdbrn", 'TXT') request = ( f"POST /dns-query HTTP/1.1\r\n" f"Host: hnsdoh.com\r\n" "Content-Type: application/dns-message\r\n" f"Content-Length: {len(dns_query)}\r\n" "Connection: close\r\n" "\r\n" ) wireframe_request = request.encode() + dns_query sock = socket.create_connection((ip, 443)) context = ssl.create_default_context() ssock = context.wrap_socket(sock, server_hostname="hnsdoh.com") try: ssock.sendall(wireframe_request) response_data = b"" while True: data = ssock.recv(4096) if not data: break response_data += data response_str = response_data.decode('latin-1') headers, body = response_str.split("\r\n\r\n", 1) dns_response:dnslib.DNSRecord = dnslib.DNSRecord.parse(body.encode('latin-1')) for rr in dns_response.rr: if "Test 2" in str(rr): status = True except Exception as e: print(e) finally: # Close the socket connection ssock.close() return status def check_dot(ip: str) -> bool: qname = dns.name.from_text("3.wdbrn") q = dns.message.make_query(qname, dns.rdatatype.TXT) try: response = dns.query.tls(q, ip, timeout=5, port=853, server_hostname="hnsdoh.com") if response.rcode() == dns.rcode.NOERROR: for rrset in response.answer: for rr in rrset: if "Test 3" in rr.to_text(): return True return False except Exception as e: print(e) return False def verify_cert(ip: str,port:int) -> bool: sock = socket.create_connection((ip, port)) # Wrap the socket in SSL/TLS context = ssl.create_default_context() ssock = context.wrap_socket(sock, server_hostname="hnsdoh.com") expires = "ERROR" try: # Retrieve the server's certificate cert = ssock.getpeercert() # Extract the expiry date from the certificate expiry_date_str = cert['notAfter'] # Convert the expiry date string to a datetime object expiry_date = datetime.strptime(expiry_date_str, '%b %d %H:%M:%S %Y GMT') expires = format_relative_time(expiry_date) valid = expiry_date > datetime.now() finally: # Close the SSL and socket connection ssock.close() return {"valid": valid, "expires": expires, "expiry_date": expiry_date_str} def format_relative_time(expiry_date: datetime) -> str: now = datetime.now() delta = expiry_date - now if delta.days > 0: return f"in {delta.days} days" if delta.days > 1 else "in 1 day" elif delta.days < 0: return f"{-delta.days} days ago" if -delta.days > 1 else "1 day ago" elif delta.seconds >= 3600: hours = delta.seconds // 3600 return f"in {hours} hours" if hours > 1 else "in 1 hour" elif delta.seconds >= 60: minutes = delta.seconds // 60 return f"in {minutes} minutes" if minutes > 1 else "in 1 minute" else: return f"in {delta.seconds} seconds" if delta.seconds > 1 else "in 1 second" def check_nodes() -> list: global nodes if last_log > datetime.now() - relativedelta.relativedelta(minutes=1): # Load the last log with open(f"{log_dir}/node_status.json", "r") as file: data = json.load(file) newest = {"date": datetime.now()-relativedelta.relativedelta(years=1), "nodes": []} for entry in data: if datetime.strptime(entry["date"], "%Y-%m-%d %H:%M:%S") > newest["date"]: newest = entry newest["date"] = datetime.strptime(newest["date"], "%Y-%m-%d %H:%M:%S") node_status = newest["nodes"] else: if len(nodes) == 0: nodes = get_node_list() node_status = [] for ip in nodes: node_status.append({ "ip": ip, "name": node_names[ip] if ip in node_names else ip, "location": node_locations[ip] if ip in node_locations else "Unknown", "plain_dns": check_plain_dns(ip), "doh": check_doh(ip), "dot": check_dot(ip), "cert": verify_cert(ip,443), "cert_853": verify_cert(ip,853) }) else: node_status = [] for ip in nodes: node_status.append({ "ip": ip, "name": node_names[ip] if ip in node_names else ip, "location": node_locations[ip] if ip in node_locations else "Unknown", "plain_dns": check_plain_dns(ip), "doh": check_doh(ip), "dot": check_dot(ip), "cert": verify_cert(ip,443), "cert_853": verify_cert(ip,853) }) # Save the node status to a file log_status(node_status) print("Finished checking nodes", flush=True) return node_status # endregion # region File logs def log_status(node_status: list): global last_log last_log = datetime.now() # Check if the file exists filename = f"{log_dir}/node_status.json" if os.path.isfile(filename): with open(filename, "r") as file: data = json.load(file) else: data = [] # Get oldest date oldest = datetime.now() newest = datetime.now()-relativedelta.relativedelta(years=1) for entry in data: date = datetime.strptime(entry["date"], "%Y-%m-%d %H:%M:%S") if date < oldest: oldest = date if date > newest: newest = date # If the oldest date is more than 7 days ago, save the file and create a new one if (datetime.now() - oldest).days > 7: # Copy the file to a new one new_filename = f"{log_dir}/node_status_{newest.strftime('%Y-%m-%d')}.json" os.rename(filename, new_filename) data = [] # Add the new entry data.append({ "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "nodes": node_status }) with open(filename, "w") as file: json.dump(data, file, indent=4) # endregion # region History functions def get_history(days: int) -> list: log_files = [f for f in os.listdir(log_dir) if f.endswith('.json') and f.startswith('node_status')] history = [] for log_file in log_files: file_path = os.path.join(log_dir, log_file) with open(file_path, 'r') as file: data = json.load(file) for entry in data: entry_date = datetime.strptime(entry["date"], "%Y-%m-%d %H:%M:%S") if datetime.now() - relativedelta.relativedelta(days=days) < entry_date: history.append(entry) return history def summarize_history(history: list) -> dict: nodes_status = defaultdict(lambda: { "name": "", "location": "", "plain_dns": {"last_down": "Never", "percentage": 0}, "doh": {"last_down": "Never", "percentage": 0}, "dot": {"last_down": "Never", "percentage": 0} }) overall_status = { "plain_dns": {"last_down": "Never", "percentage": 0}, "doh": {"last_down": "Never", "percentage": 0}, "dot": {"last_down": "Never", "percentage": 0} } # Collect data total_counts = defaultdict(lambda: { "plain_dns": {"down": 0, "total": 0}, "doh": {"down": 0, "total": 0}, "dot": {"down": 0, "total": 0} }) for entry in history: date = datetime.strptime(entry["date"], "%Y-%m-%d %H:%M:%S") for node in entry["nodes"]: ip = node["ip"] # Update node details if not already present if nodes_status[ip]["name"] == "": nodes_status[ip]["name"] = node.get("name", "") nodes_status[ip]["location"] = node.get("location", "") # Update counts and last downtime for key in ["plain_dns", "doh", "dot"]: status = node.get(key, "up") if status == "down": total_counts[ip][key]["down"] += 1 total_counts[ip][key]["total"] += 1 # Update last downtime for each key for key in ["plain_dns", "doh", "dot"]: if node.get(key) == "down": nodes_status[ip][key]["last_down"] = date.strftime("%Y-%m-%d %H:%M:%S") # Calculate percentages and prepare final summary node_list = [] for ip, status in nodes_status.items(): node_data = status.copy() for key in ["plain_dns", "doh", "dot"]: total = total_counts[ip][key]["total"] down = total_counts[ip][key]["down"] if total > 0: node_data[key]["percentage"] = ((total - down) / total) * 100 else: node_data[key]["percentage"] = 100 node_list.append(node_data) # Aggregate overall status overall_counts = {"plain_dns": {"down": 0, "total": 0}, "doh": {"down": 0, "total": 0}, "dot": {"down": 0, "total": 0}} for ip, counts in total_counts.items(): for key in ["plain_dns", "doh", "dot"]: overall_counts[key]["total"] += counts[key]["total"] overall_counts[key]["down"] += counts[key]["down"] for key in ["plain_dns", "doh", "dot"]: total = overall_counts[key]["total"] down = overall_counts[key]["down"] if total > 0: overall_status[key]["percentage"] = ((total - down) / total) * 100 last_downs = [nodes_status[ip][key]["last_down"] for ip in nodes_status if nodes_status[ip][key]["last_down"] != "Never"] if last_downs: overall_status[key]["last_down"] = max(last_downs) else: overall_status[key]["percentage"] = 100 return { "nodes": node_list, "overall": overall_status, "check_counts": total_counts } def convert_nodes_to_dict(nodes): nodes_dict = {} for node in nodes: name = node.get('name') if name: nodes_dict[name] = node return nodes_dict # endregion # region API routes @app.route("/api/nodes") def api_nodes(): node_status = check_nodes() return jsonify(node_status) @app.route("/api/history") def api_history(): history_days = 7 if "days" in request.args: try: history_days = int(request.args["days"]) except: pass history = get_history(history_days) history_summary = summarize_history(history) return jsonify(history_summary) # endregion # region Main routes @app.route("/") def index(): node_status = check_nodes() warnings = [] for node in node_status: if not node["plain_dns"]: warnings.append(f"{node['name']} does not support plain DNS") if not node["doh"]: warnings.append(f"{node['name']} does not support DoH") if not node["dot"]: warnings.append(f"{node['name']} does not support DoT") if not node["cert"]["valid"]: warnings.append(f"{node['name']} has an invalid certificate") if not node["cert_853"]["valid"]: warnings.append(f"{node['name']} has an invalid certificate on port 853") cert_expiry = datetime.strptime(node["cert"]["expiry_date"], '%b %d %H:%M:%S %Y GMT') if cert_expiry < datetime.now() + relativedelta.relativedelta(days=7): warnings.append(f"{node['name']} has a certificate expiring in less than 7 days on port 443") cert_853_expiry = datetime.strptime(node["cert_853"]["expiry_date"], '%b %d %H:%M:%S %Y GMT') if cert_853_expiry < datetime.now() + relativedelta.relativedelta(days=7): warnings.append(f"{node['name']} has a certificate expiring in less than 7 days on port 853") history_days = 7 if "history" in request.args: try: history_days = int(request.args["history"]) except: pass history = get_history(history_days) history_summary = summarize_history(history) history_summary["nodes"] = convert_nodes_to_dict(history_summary["nodes"]) last_check = last_log.strftime("%Y-%m-%d %H:%M:%S") print(f"Last check: {last_check}", flush=True) return render_template("index.html",nodes=node_status,warnings=warnings,history=history_summary,last_check=last_check) @app.route("/") def catch_all(path: str): if os.path.isfile("templates/" + path): return render_template(path) # Try with .html if os.path.isfile("templates/" + path + ".html"): return render_template(path + ".html") if os.path.isfile("templates/" + path.strip("/") + ".html"): return render_template(path.strip("/") + ".html") # Try to find a file matching if path.count("/") < 1: # Try to find a file matching filename = find(path, "templates") if filename: return send_file(filename) return render_template("404.html"), 404 # endregion # region Error Catching # 404 catch all @app.errorhandler(404) def not_found(e): return render_template("404.html"), 404 # endregion if __name__ == "__main__": app.run(debug=True, port=5000, host="0.0.0.0")