app_dash.py

import requests
import time
import subprocess
from datetime import datetime
import curses
import dnsstamps
from dnsstamps import Option
import numpy as np
import json
import threading

# IPFS API URL
IPFS_API_URL = "http://localhost:5001/api/v0"

# DNSCrypt-proxy configuration path
DNSCRYPT_CONFIG_PATH = "/etc/dnscrypt-proxy/dnscrypt-proxy.toml"

# Global variable to store current status
current_status = {
    "ip_address": "Not resolved",
    "last_ip_address": None,
    "connectivity": "Unknown",
    "last_update": "Never",
    "latency": [],
    "average_latency": "N/A",
    "median_latency": "N/A",
    "p90_latency": "N/A",
    "dnscrypt_reachable": False,
    "doh_reachable": False,
    "ipfs_reachable": False,
    "current_action": "Running...",
}

# Global list to store error logs
error_logs = []

# Global variable to store IPNS key
ipns_key = ""


def log_error(message):
    """
    Log an error message to the error_logs list, maintaining only the last 5 entries.
    """
    if len(error_logs) >= 5:
        error_logs.pop(0)
    error_logs.append(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {message}")


def get_latest_ipns_content():
    """
    Resolve the latest IPFS content using the IPNS key with no cache.
    """
    current_status["current_action"] = "Resolving IPNS..."
    try:
        response = requests.post(f"{IPFS_API_URL}/name/resolve", params={"arg": ipns_key, "nocache": "true"})
        response.raise_for_status()
        cid = response.json().get('Path')
        current_status["ipfs_reachable"] = True
        return cid
    except requests.exceptions.RequestException as e:
        current_status["ipfs_reachable"] = False
        log_error(f"Error resolving IPNS: {str(e)}")
        return None


def get_ip_from_ipfs(cid):
    """
    Retrieve the IP address from the IPFS content using the IPFS API.
    """
    current_status["current_action"] = "Fetching IP address from IPFS..."
    try:
        response = requests.post(f"{IPFS_API_URL}/cat", params={"arg": cid})
        response.raise_for_status()

        try:
            content = response.json()
            log_error(f"content from ipfs: {str(content)}")
        except json.JSONDecodeError:
            log_error("Failed to decode IPFS response as JSON.")
            return None, None

        # Extract IP and query path
        ip_address = content.get('ip')
        query_path = content.get('query_path', "NOT WORKING")

        # Log the values for debugging
        log_error(f"IP Address retrieved: {ip_address}, Query Path retrieved: {query_path}")
        
        return ip_address, query_path

    except requests.exceptions.RequestException as e:
        log_error(f"Error retrieving IP from IPFS: {str(e)}")
        return None, None


def create_dnscrypt_stamp(ip_address, hostname, query_path):
    """
    Create a DNSCrypt DoH stamp based on the server's IP address and hostname.
    """
    try:
        stamp = dnsstamps.create_doh(ip_address, [], hostname, query_path, [Option.NO_LOGS, Option.DNSSEC, Option.NO_FILTERS])
        return stamp
    except Exception as e:
        log_error(f"Error creating DNSCrypt stamp: {str(e)}")
        return None


def update_dnscrypt_config(ip_address, hostname, query_path):
    """
    Update the dnscrypt-proxy configuration with the new IP address and stamp.
    """
    current_status["current_action"] = "Updating DNSCrypt-proxy configuration..."
    stamp = create_dnscrypt_stamp(ip_address, hostname, query_path)
    if not stamp:
        return

    try:
        with open(DNSCRYPT_CONFIG_PATH, 'r') as file:
            config = file.readlines()

        with open(DNSCRYPT_CONFIG_PATH, 'w') as file:
            for line in config:
                if line.strip().startswith('stamp ='):
                    file.write(f"stamp = '{stamp}'\n")
                else:
                    file.write(line)
    except FileNotFoundError:
        log_error(f"Configuration file {DNSCRYPT_CONFIG_PATH} not found.")
    except Exception as e:
        log_error(f"Error updating DNSCrypt-proxy configuration: {str(e)}")


def restart_dnscrypt_proxy():
    """
    Restart the dnscrypt-proxy service.
    """
    current_status["current_action"] = "Restarting DNSCrypt-proxy service..."
    try:
        subprocess.run(['sudo', 'systemctl', 'restart', 'dnscrypt-proxy'], check=True)
        time.sleep(5)
    except subprocess.CalledProcessError as e:
        log_error(f"Error restarting DNSCrypt-proxy: {str(e)}")


def test_doh_server(ip_address):
    """
    Test if the DoH server is reachable at the new IP address using dig and measure latency.
    """
    current_status["current_action"] = "Testing DoH server connectivity..."
    try:
        start_time = time.time()
        result = subprocess.run(['dig', '@127.0.0.1', 'google.com'], capture_output=True, text=True)
        end_time = time.time()
        latency = (end_time - start_time) * 1000  # Convert to milliseconds

        if result.returncode == 0:
            current_status["latency"].append(latency)
            if len(current_status["latency"]) > 20:  # Keep only the last 20 latency measurements
                current_status["latency"].pop(0)
            update_statistics()
            current_status["dnscrypt_reachable"] = True
            current_status["doh_reachable"] = True  # Assuming if the localhost query works, the upstream DoH works too.
            return "Reachable"
        else:
            current_status["dnscrypt_reachable"] = False
            current_status["doh_reachable"] = False
            log_error(f"DoH server unreachable: {result.stderr.strip()}")
            return "Unreachable"
    except Exception as e:
        current_status["dnscrypt_reachable"] = False
        current_status["doh_reachable"] = False
        log_error(f"Error testing DoH server: {str(e)}")
        return "Error"


def update_statistics():
    """
    Update average, median, and 90th percentile latency statistics.
    """
    if current_status["latency"]:
        latencies = np.array(current_status["latency"])
        current_status["average_latency"] = f"{np.mean(latencies):.2f} ms"
        current_status["median_latency"] = f"{np.median(latencies):.2f} ms"
        current_status["p90_latency"] = f"{np.percentile(latencies, 90):.2f} ms"
    else:
        current_status["average_latency"] = "N/A"
        current_status["median_latency"] = "N/A"
        current_status["p90_latency"] = "N/A"


def update_status():
    """
    Update the current status by checking the IPNS content, updating the dnscrypt-proxy,
    and testing connectivity if the IP address has changed.
    """
    cid = get_latest_ipns_content()
    if cid:
        ip_address, query_path = get_ip_from_ipfs(cid)
        hostname = ip_address + ":443" if ip_address else "Unknown"
        if ip_address and ip_address != current_status["last_ip_address"]:
            # Update configuration and restart service only if the IP has changed
            update_dnscrypt_config(ip_address, hostname, query_path)
            restart_dnscrypt_proxy()
            current_status["last_ip_address"] = ip_address

        # Test connectivity whether the IP has changed or not
        connectivity = test_doh_server(ip_address)
        current_status["ip_address"] = ip_address
        current_status["connectivity"] = connectivity
        current_status["last_update"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        current_status["current_action"] = "Waiting for update check..."


def listen_for_ipns_updates():
    """
    Listens for updates on the IPNS pubsub topic. When an update is received,
    it triggers an IPNS resolve and processes the content.
    """
    pubsub_topic = f"/ipns/{ipns_key}"
    
    try:
        # Start the IPFS pubsub sub command
        process = subprocess.Popen(
            ['ipfs', 'pubsub', 'sub', pubsub_topic],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        # Listen for messages from the pubsub topic
        while True:
            output = process.stdout.readline().decode('utf-8').strip()
            if output:
                log_error(f"Received pubsub message: {output}")
                
                # Trigger IPNS resolve and process the content
                cid = get_latest_ipns_content()
                if cid:
                    ip_address, query_path = get_ip_from_ipfs(cid)
                    hostname = ip_address + ":443" if ip_address else "Unknown"
                    if ip_address and ip_address != current_status["last_ip_address"]:
                        # Update configuration and restart service only if the IP has changed
                        update_dnscrypt_config(ip_address, hostname, query_path)
                        restart_dnscrypt_proxy()
                        current_status["last_ip_address"] = ip_address

                    # Test connectivity whether the IP has changed or not
                    connectivity = test_doh_server(ip_address)
                    current_status["ip_address"] = ip_address
                    current_status["connectivity"] = connectivity
                    current_status["last_update"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    current_status["current_action"] = "Waiting for update check..."

    except Exception as e:
        log_error(f"Error in pubsub listener: {str(e)}")


def draw_status_indicator(stdscr, y, x, status):
    """
    Draw a colored circle based on the status.
    Green circle for reachable, red for unreachable.
    """
    if status:
        stdscr.addstr(y, x, "●", curses.color_pair(1))  # Green
    else:
        stdscr.addstr(y, x, "●", curses.color_pair(2))  # Red


def draw_statistics_table(stdscr):
    """
    Draw a table with the statistics on the screen.
    """
    stdscr.addstr(18, 2, "Latency Statistics")
    stdscr.addstr(20, 2, f"{'Metric':<20}{'Value':<10}", curses.A_UNDERLINE)
    stdscr.addstr(21, 2, f"{'Average':<20}{current_status['average_latency']:<10}")
    stdscr.addstr(22, 2, f"{'Median':<20}{current_status['median_latency']:<10}")
    stdscr.addstr(23, 2, f"{'90th Percentile':<20}{current_status['p90_latency']:<10}")


def draw_error_logs(stdscr):
    """
    Draw error logs at the bottom of the screen.
    """
    stdscr.addstr(23, 2, "Error Logs:")
    for i, log in enumerate(error_logs[-5:]):  # Display last 5 error logs
        stdscr.addstr(24 + i, 2, log)


def draw_dashboard(stdscr):
    """
    Draw the terminal dashboard using curses.
    """
    curses.curs_set(0)
    curses.start_color()
    curses.init_pair(1, curses.COLOR_GREEN, curses.COLOR_BLACK)  # Green
    curses.init_pair(2, curses.COLOR_RED, curses.COLOR_BLACK)    # Red
    stdscr.nodelay(1)
    stdscr.timeout(1000)

    # ASCII art title and subtitle
    title = [
        " _   _ _        _       ____        _   _ ",
        "| \\ | (_)_ __  (_) __ _|  _ \\  ___ | | | |",
        "|  \\| | | '_ \\ | |/ _` | | | |/ _ \\| |_| |",
        "| |\\  | | | | || | (_| | |_| | (_) |  _  |",
        "|_| \\_|_|_| |_|/ |\\__,_|____/ \\___/|_| |_|",
        "             |__/                         ",
    ]
    subtitle = "by Secretlab"

    while True:
        stdscr.clear()

        # Draw the title and subtitle
        for i, line in enumerate(title):
            stdscr.addstr(i, 2, line, curses.A_BOLD)
        stdscr.addstr(len(title), 4, subtitle, curses.A_DIM)

        # Draw the header
        stdscr.addstr(10, 2, f"Current IP Address: {current_status['ip_address']}")
        stdscr.addstr(11, 2, f"Connectivity Status: {current_status['connectivity']}")
        stdscr.addstr(12, 2, f"Current Action: {current_status['current_action']}")

        # Draw connectivity indicators
        stdscr.addstr(14, 2, "DNSCrypt-Proxy Status: ")
        draw_status_indicator(stdscr, 14, 25, current_status["dnscrypt_reachable"])
        stdscr.addstr(15, 2, "DoH Server Status: ")
        draw_status_indicator(stdscr, 15, 25, current_status["doh_reachable"])
        stdscr.addstr(16, 2, "IPFS Node Status: ")
        draw_status_indicator(stdscr, 16, 25, current_status["ipfs_reachable"])

        # Draw the latency statistics table
        draw_statistics_table(stdscr)

        # Draw error logs
        draw_error_logs(stdscr)

        # Update status every 5 seconds for testing
        if datetime.now().second % 5 == 0:
            update_status()

        stdscr.refresh()

        key = stdscr.getch()
        if key == ord('q'):
            break


def main():
    global ipns_key
    # Prompt user for the IPNS key
    ipns_key = input("Enter the IPNS key to use: ")

    # Start the IPNS pubsub listener in a separate thread
    pubsub_thread = threading.Thread(target=listen_for_ipns_updates, daemon=True)
    pubsub_thread.start()

    curses.wrapper(draw_dashboard)


if __name__ == "__main__":
    main()