Launch from Canaveral

"""

Canaveral, a simple Gemini server

Key features:
- Recursively host a static directory via Gemini
- Concurrency
- Certificate management 
- Automatic MIME types
- Argument parser for ease of use
- Minimal dependencies (standard library & openssl)
- Under 200 lines of Python

"""

import os
import ssl
import sys
import argparse
import mimetypes
import threading
import subprocess
import socket
from pathlib import Path
from urllib.parse import urlparse, unquote

LOGO = r"""
   ██████╗ █████╗ ███╗   ██╗ █████╗ ██╗   ██╗███████╗██████╗  █████╗ ██╗     
  ██╔════╝██╔══██╗████╗  ██║██╔══██╗██║   ██║██╔════╝██╔══██╗██╔══██╗██║     
  ██║     ███████║██╔██╗ ██║███████║██║   ██║█████╗  ██████╔╝███████║██║     
  ██║     ██╔══██║██║╚██╗██║██╔══██║╚██╗ ██╔╝██╔══╝  ██╔══██╗██╔══██║██║     
  ╚██████╗██║  ██║██║ ╚████║██║  ██║ ╚████╔╝ ███████╗██║  ██║██║  ██║███████╗
   ╚═════╝╚═╝  ╚═╝╚═╝  ╚═══╝╚═╝  ╚═╝  ╚═══╝  ╚══════╝╚═╝  ╚═╝╚═╝  ╚═╝╚══════╝
"""

parser = argparse.ArgumentParser()
parser.add_argument("-g", "--generate-cert", help="generate certificate", action="store_true")
parser.add_argument("-c", "--cert", type=str, help="path to certificate file", default="cert.pem")
parser.add_argument("-k", "--key", type=str, help="path to key file", default="key.pem")
parser.add_argument("-C", "--cn", type=str, help="common name", default="localhost")
parser.add_argument("-s", "--san", type=str, help="subject alternative names", default="localhost,127.0.0.1")
parser.add_argument("-D", "--days", type=int, help="certificate validity period", default=365)
parser.add_argument("-p", "--port", type=int, help="server port", default=1965)
parser.add_argument("-H", "--host", type=str, help="server host", default="0.0.0.0")
parser.add_argument("-d", "--dir", type=str, help="server root", default="~/public_gemini")
args = parser.parse_args()

CERT = Path(args.cert)
KEY = Path(args.key)
SERVER_ROOT = Path(args.dir).expanduser().resolve()

def check_cert_valid(cert_path):
    if not cert_path.exists():
        return False

    try:
        result = subprocess.run(["openssl", "x509", "-checkend", "0", "-noout", "-in", str(cert_path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return result.returncode == 0
    except FileNotFoundError:
        print("OpenSSL not found in PATH.")
        sys.exit(1)

def build_san(san_str):
    entries = []
    for item in san_str.split(","):
        item = item.strip()
        if not item:
            continue
        if item.replace(".", "").isdigit():  # naive IP check
            entries.append(f"IP:{item}")
        else:
            entries.append(f"DNS:{item}")
    return "subjectAltName=" + ",".join(entries)

def generate_cert():
    print("Generating new self-signed certificate...")
    san = build_san(args.san)
    subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-keyout", str(KEY), "-out", str(CERT), "-days", str(args.days), "-nodes", "-subj", f"/CN={args.cn}", "-addext", san], check=True)

def ensure_cert():
    if args.generate_cert:
        generate_cert()
        return

    if (CERT.exists() and KEY.exists() and check_cert_valid(CERT)):
        print("Certificate is valid.")
        return

    confirm = input("No SSL certificate was detected. Generate one? (y/n): ").strip().lower()
    if confirm != "y":
        print("Cannot proceed without certificate.")
        sys.exit(1)

    generate_cert()

def resolve_path(request_bytes):
    url = request_bytes.decode("utf-8", "ignore").strip()
    parsed = urlparse(url)
    raw_path = unquote(parsed.path)

    rel = raw_path.lstrip("/")
    candidate = (SERVER_ROOT / rel).resolve()

    if not candidate.is_relative_to(SERVER_ROOT):
        raise PermissionError("Path escapes server root")

    if candidate.exists() and candidate.is_dir():
        candidate = candidate / "index.gmi"

    return candidate

mimetypes.add_type("text/gemini", ".gmi")
mimetypes.add_type("text/gemini", ".gemini")

def guess_mime(path):
    mime, encoding = mimetypes.guess_type(path)
    if mime is None:
        return "application/octet-stream"
    if mime.startswith("text/"):
        return f"{mime}; charset=utf-8" # specify utf-8 for text data
    return mime

shutdown_event = threading.Event()
threads = []

def handle_client(conn, addr, context):
    try:
        with context.wrap_socket(conn, server_side=True) as tls:
            tls.settimeout(10)

            request = tls.recv(1024)

            try:
                path = resolve_path(request)

                if not path.exists() or not path.is_file():
                    tls.sendall(b"51 Not Found\r\n")
                    return

                send_success(tls, path)

            except PermissionError:
                tls.sendall(b"53 Proxy Request Refused\r\n")
            except ValueError:
                tls.sendall(b"59 Bad Request\r\n")

    except socket.timeout:
        print(f"Connection timed out from {addr}")
    except ssl.SSLError as e:
        print(f"TLS failed from {addr}: {e}")
    finally:
        conn.close()

def exec_server():
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(CERT, KEY)

    with socket.socket() as sock:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind((args.host, args.port))
        sock.listen()
        sock.settimeout(1.0)

        print(f"Preparing to serve {SERVER_ROOT}.")
        print(f"Listening on {args.port}...")

        try:
            while not shutdown_event.is_set():
                try:
                    conn, addr = sock.accept()
                except socket.timeout:
                    continue

                thread = threading.Thread(target=handle_client, args=(conn, addr, context))
                thread.start()
                threads.append(thread)

        except KeyboardInterrupt:
            print("\nInterrupt received.")
            shutdown_event.set()

        finally:
            if len(threads) > 0:
                print("Grace period for active connections to finish...")

            for thread in threads:
                thread.join(timeout=2)

def send_success(conn, path):
    header = f"20 {guess_mime(path)}\r\n".encode("utf-8")
    conn.sendall(header)

    # send chunks
    with path.open("rb") as f:
        while chunk := f.read(8192):
            conn.sendall(chunk)

def main():
    print(LOGO)
    ensure_cert()
    exec_server()

if __name__ == "__main__":
    main()