box/bin/crypt-server.py

#!/usr/bin/env python3
"""
crypt-server.py — Public cryptographic attestation and key directory for NetVM.

Serves https://crypt.muse-dev.online/ (via cloudflared / reverse proxy).
Endpoints:
  GET /                         -> Service directory / health JSON
  GET /health                   -> Health check
  GET /keys/allowed_signers     -> OpenSSH allowed_signers formatted file
  GET /keys/{identity}.pub      -> Individual public key
  GET /proofs                   -> List known proof hashes / work order attestations
  GET /proofs/{id}              -> Retrieve proof envelope and SSH signature
  POST /proofs                  -> Submit / register a signed proof record
"""

import argparse
import glob
import json
import os
import re
import ssl
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler

REPO_DIR = "/home/super/Projects/NetVM"
SIGNERS_DIR = os.path.join(REPO_DIR, "dm-signers")
PROOFS_DIR = os.path.join(REPO_DIR, "var", "proofs")

os.makedirs(PROOFS_DIR, exist_ok=True)

class CryptHandler(BaseHTTPRequestHandler):
    server_version = "crypt-attestation/1.0"

    def log_message(self, format, *args):
        sys.stderr.write(f"crypt-server: {self.client_address[0]} - {format % args}\n")

    def _send(self, code, content, content_type="application/json"):
        if isinstance(content, (dict, list)):
            body = json.dumps(content, indent=2).encode("utf-8")
        elif isinstance(content, str):
            body = content.encode("utf-8")
        else:
            body = bytes(content)

        self.send_response(code)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = self.path.split("?")[0].rstrip("/")
        if not path:
            path = "/"

        if path in ("/", "/health"):
            self._send(200, {
                "service": "crypt.muse-dev.online",
                "status": "active",
                "mode": "public_attestation",
                "endpoints": [
                    "/keys/allowed_signers",
                    "/keys/<identity>.pub",
                    "/proofs",
                    "/proofs/<id>"
                ]
            })
            return

        if path == "/keys/allowed_signers":
            allowed_path = os.path.join(SIGNERS_DIR, "allowed_signers")
            if os.path.exists(allowed_path):
                with open(allowed_path, "r") as f:
                    data = f.read()
                self._send(200, data, content_type="text/plain; charset=utf-8")
            else:
                self._send(404, {"error": "allowed_signers not found"})
            return

        m_key = re.match(r"^/keys/([a-zA-Z0-9_\-\.]+)\.pub$", path)
        if m_key:
            ident = m_key.group(1)
            pub_path = os.path.join(SIGNERS_DIR, f"{ident}.pub")
            if os.path.exists(pub_path):
                with open(pub_path, "r") as f:
                    data = f.read()
                self._send(200, data, content_type="text/plain; charset=utf-8")
            else:
                self._send(404, {"error": f"Public key for {ident} not found"})
            return

        if path == "/proofs":
            proof_files = glob.glob(os.path.join(PROOFS_DIR, "*.json"))
            ids = [os.path.basename(p)[:-5] for p in proof_files]
            self._send(200, {"proofs": sorted(ids)})
            return

        m_proof = re.match(r"^/proofs/([a-zA-Z0-9_\-]+)$", path)
        if m_proof:
            pid = m_proof.group(1)
            pf = os.path.join(PROOFS_DIR, f"{pid}.json")
            if os.path.exists(pf):
                with open(pf, "r") as f:
                    data = json.load(f)
                self._send(200, data)
            else:
                self._send(404, {"error": f"Proof {pid} not found"})
            return

        self._send(404, {"error": "not found"})

    def do_POST(self):
        path = self.path.split("?")[0].rstrip("/")
        if path == "/proofs":
            try:
                length = int(self.headers.get("Content-Length", 0))
            except ValueError:
                length = 0
            if length <= 0 or length > 65536:
                self._send(400, {"error": "invalid content length"})
                return
            try:
                data = json.loads(self.rfile.read(length))
            except Exception:
                self._send(400, {"error": "malformed JSON"})
                return

            pid = data.get("id")
            if not pid or not re.match(r"^[a-zA-Z0-9_\-]+$", pid):
                self._send(400, {"error": "missing or invalid proof id"})
                return

            pf = os.path.join(PROOFS_DIR, f"{pid}.json")
            with open(pf, "w") as f:
                json.dump(data, f, indent=2)

            self._send(201, {"status": "stored", "id": pid, "url": f"https://crypt.muse-dev.online/proofs/{pid}"})
            return

        self._send(404, {"error": "not found"})


def main():
    parser = argparse.ArgumentParser(description="Crypt attestation and key server")
    parser.add_argument("--port", type=int, default=8446)
    parser.add_argument("--host", default="100.123.153.75")
    parser.add_argument("--ssl", action="store_true", help="Enable self-signed HTTPS")
    args = parser.parse_args()

    server = HTTPServer((args.host, args.port), CryptHandler)

    if args.ssl:
        cert_file = os.path.join(REPO_DIR, "ssl", "crypt-selfsigned.crt")
        key_file = os.path.join(REPO_DIR, "ssl", "crypt-selfsigned.key")
        os.makedirs(os.path.dirname(cert_file), exist_ok=True)
        if not os.path.exists(cert_file):
            import subprocess
            subprocess.run([
                "openssl", "req", "-x509", "-newkey", "rsa:2048",
                "-keyout", key_file, "-out", cert_file,
                "-days", "3650", "-nodes",
                "-subj", "/CN=crypt.muse-dev.online"
            ], check=True, capture_output=True)
            os.chmod(key_file, 0o600)
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_cert_chain(cert_file, key_file)
        server.socket = ctx.wrap_socket(server.socket, server_side=True)

    print(f"Crypt server running on {'https' if args.ssl else 'http'}://{args.host}:{args.port}", file=sys.stderr)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down crypt server", file=sys.stderr)


if __name__ == "__main__":
    main()