api.py

#!/usr/bin/env python3
"""UptimePulse API — HTTP server for managing monitors."""

import json
import sqlite3
import os
import sys
import uuid
import time
import threading
import logging
from http.server import HTTPServer, BaseHTTPRequestHandler
from datetime import datetime, timezone, timedelta
from urllib.parse import urlparse, parse_qs

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "uptimepulse.db")
MONITORS_PATH = os.path.join(BASE_DIR, "monitors.json")
LOG_PATH = os.path.join(BASE_DIR, "api.log")
PORT = 8090
MAX_MONITORS = 100
RATE_LIMIT = 10  # requests per minute per IP
BASE_URL = "https://65-109-129-230.sslip.io/pulse"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(LOG_PATH),
        logging.StreamHandler(sys.stdout),
    ],
)
log = logging.getLogger("uptimepulse-api")

# Rate limiting storage: {ip: [timestamps]}
rate_limits = {}
rate_lock = threading.Lock()


def check_rate_limit(ip):
    """Returns True if request is allowed, False if rate limited."""
    now = time.time()
    with rate_lock:
        if ip not in rate_limits:
            rate_limits[ip] = []
        # Remove entries older than 60 seconds
        rate_limits[ip] = [t for t in rate_limits[ip] if now - t < 60]
        if len(rate_limits[ip]) >= RATE_LIMIT:
            return False
        rate_limits[ip].append(now)
        return True


def load_monitors():
    try:
        with open(MONITORS_PATH, "r") as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {"monitors": []}


def save_monitors(data):
    with open(MONITORS_PATH, "w") as f:
        json.dump(data, f, indent=2)


def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def generate_badge_svg(status):
    """Generate a shields.io-style SVG badge."""
    if status == "up":
        color = "#22c55e"
        text = "up"
        text_width = 18
    elif status == "down":
        color = "#ef4444"
        text = "down"
        text_width = 30
    else:
        color = "#94a3b8"
        text = "unknown"
        text_width = 50

    label = "status"
    label_width = 40
    total_width = label_width + text_width + 20

    svg = f"""<svg xmlns="http://www.w3.org/2000/svg" width="{total_width}" height="20" role="img">
  <linearGradient id="s" x2="0" y2="100%">
    <stop offset="0" stop-color="#bbb" stop-opacity=".1"/>
    <stop offset="1" stop-opacity=".1"/>
  </linearGradient>
  <clipPath id="r"><rect width="{total_width}" height="20" rx="3" fill="#fff"/></clipPath>
  <g clip-path="url(#r)">
    <rect width="{label_width + 10}" height="20" fill="#555"/>
    <rect x="{label_width + 10}" width="{text_width + 10}" height="20" fill="{color}"/>
    <rect width="{total_width}" height="20" fill="url(#s)"/>
  </g>
  <g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,sans-serif" font-size="11">
    <text x="{(label_width + 10) / 2}" y="15" fill="#010101" fill-opacity=".3">{label}</text>
    <text x="{(label_width + 10) / 2}" y="14">{label}</text>
    <text x="{label_width + 10 + (text_width + 10) / 2}" y="15" fill="#010101" fill-opacity=".3">{text}</text>
    <text x="{label_width + 10 + (text_width + 10) / 2}" y="14">{text}</text>
  </g>
</svg>"""
    return svg


class UptimePulseHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        log.info(f"{self.client_address[0]} - {format % args}")

    def send_json(self, code, data):
        body = json.dumps(data, indent=2).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()
        self.wfile.write(body)

    def send_svg(self, svg):
        body = svg.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "image/svg+xml")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def get_client_ip(self):
        forwarded = self.headers.get("X-Forwarded-For")
        if forwarded:
            return forwarded.split(",")[0].strip()
        return self.client_address[0]

    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def do_GET(self):
        ip = self.get_client_ip()
        if not check_rate_limit(ip):
            self.send_json(429, {"error": "Rate limit exceeded. Max 10 requests/minute."})
            return

        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/")

        # Strip /pulse/api prefix if present (Caddy passes full path)
        if path.startswith("/pulse/api"):
            path = path[len("/pulse"):]
        elif path.startswith("/pulse"):
            path = path[len("/pulse"):]

        if path == "/api/health":
            self.handle_health()
        elif path.startswith("/api/monitors/") and path.endswith("/status"):
            monitor_id = path.split("/")[3]
            self.handle_get_status(monitor_id)
        elif path.startswith("/api/monitors/") and path.endswith("/badge"):
            monitor_id = path.split("/")[3]
            self.handle_get_badge(monitor_id)
        else:
            self.send_json(404, {"error": "Not found"})

    def do_POST(self):
        ip = self.get_client_ip()
        if not check_rate_limit(ip):
            self.send_json(429, {"error": "Rate limit exceeded. Max 10 requests/minute."})
            return

        path = urlparse(self.path).path.rstrip("/")
        if path.startswith("/pulse/api"):
            path = path[len("/pulse"):]
        elif path.startswith("/pulse"):
            path = path[len("/pulse"):]

        if path == "/api/monitors":
            self.handle_create_monitor()
        else:
            self.send_json(404, {"error": "Not found"})

    def do_DELETE(self):
        ip = self.get_client_ip()
        if not check_rate_limit(ip):
            self.send_json(429, {"error": "Rate limit exceeded. Max 10 requests/minute."})
            return

        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/")
        if path.startswith("/pulse/api"):
            path = path[len("/pulse"):]
        elif path.startswith("/pulse"):
            path = path[len("/pulse"):]
        query = parse_qs(parsed.query)

        if path.startswith("/api/monitors/"):
            monitor_id = path.split("/")[3]
            token = query.get("token", [None])[0]
            self.handle_delete_monitor(monitor_id, token)
        else:
            self.send_json(404, {"error": "Not found"})

    def handle_health(self):
        data = load_monitors()
        self.send_json(200, {
            "status": "ok",
            "service": "UptimePulse",
            "monitors_count": len(data.get("monitors", [])),
            "max_monitors": MAX_MONITORS,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        })

    def handle_create_monitor(self):
        try:
            content_length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_length)
            params = json.loads(body)
        except (ValueError, json.JSONDecodeError):
            self.send_json(400, {"error": "Invalid JSON body"})
            return

        url = params.get("url", "").strip()
        name = params.get("name", "").strip()
        email = params.get("email", "").strip()
        webhook_url = params.get("webhook_url", "").strip()

        if not url or not name or not email:
            self.send_json(400, {"error": "url, name, and email are required"})
            return

        # Validate URL
        parsed_url = urlparse(url)
        if parsed_url.scheme not in ("http", "https"):
            self.send_json(400, {"error": "URL must start with http:// or https://"})
            return

        data = load_monitors()
        if len(data.get("monitors", [])) >= MAX_MONITORS:
            self.send_json(403, {"error": f"Maximum {MAX_MONITORS} monitors reached (free tier limit)"})
            return

        # Check for duplicate URL
        for m in data.get("monitors", []):
            if m["url"] == url and m["status"] == "active":
                self.send_json(409, {"error": "This URL is already being monitored", "id": m["id"]})
                return

        monitor_id = uuid.uuid4().hex[:12]
        delete_token = uuid.uuid4().hex

        monitor = {
            "id": monitor_id,
            "url": url,
            "name": name,
            "email": email,
            "interval_seconds": 300,
            "webhook_url": webhook_url if webhook_url else None,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "status": "active",
            "delete_token": delete_token,
        }

        data["monitors"].append(monitor)
        save_monitors(data)

        log.info(f"Monitor created: {name} ({url}) -> {monitor_id}")

        status_page_url = f"{BASE_URL}/status/{monitor_id}.html"
        badge_url = f"{BASE_URL}/api/monitors/{monitor_id}/badge"

        self.send_json(201, {
            "id": monitor_id,
            "url": url,
            "name": name,
            "status_page_url": status_page_url,
            "badge_url": badge_url,
            "delete_token": delete_token,
            "message": "Monitor created. First check will run within 5 minutes.",
        })

    def handle_get_status(self, monitor_id):
        data = load_monitors()
        monitor = None
        for m in data.get("monitors", []):
            if m["id"] == monitor_id:
                monitor = m
                break

        if not monitor:
            self.send_json(404, {"error": "Monitor not found"})
            return

        conn = get_db()
        try:
            # Latest check
            cur = conn.execute(
                "SELECT status, status_code, response_time_ms, timestamp, error FROM checks WHERE monitor_id = ? ORDER BY timestamp DESC LIMIT 1",
                (monitor_id,),
            )
            latest = cur.fetchone()

            # 24h history
            since_24h = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat()
            cur = conn.execute(
                "SELECT timestamp, status, status_code, response_time_ms FROM checks WHERE monitor_id = ? AND timestamp >= ? ORDER BY timestamp DESC",
                (monitor_id, since_24h),
            )
            history = [
                {"timestamp": row[0], "status": row[1], "status_code": row[2], "response_time_ms": row[3]}
                for row in cur.fetchall()
            ]

            # Uptime calc
            total_24h = len(history)
            up_24h = sum(1 for h in history if h["status"] == "up")
            uptime_24h = round((up_24h / total_24h * 100), 2) if total_24h > 0 else 100.0

            result = {
                "id": monitor_id,
                "url": monitor["url"],
                "name": monitor["name"],
                "status": latest[0] if latest else "unknown",
                "status_code": latest[1] if latest else None,
                "response_time_ms": latest[2] if latest else None,
                "last_checked": latest[3] if latest else None,
                "uptime_24h": uptime_24h,
                "checks_24h": total_24h,
                "history": history[:50],
            }
            self.send_json(200, result)
        finally:
            conn.close()

    def handle_get_badge(self, monitor_id):
        conn = get_db()
        try:
            cur = conn.execute(
                "SELECT status FROM checks WHERE monitor_id = ? ORDER BY timestamp DESC LIMIT 1",
                (monitor_id,),
            )
            row = cur.fetchone()
            status = row[0] if row else "unknown"
        finally:
            conn.close()

        svg = generate_badge_svg(status)
        self.send_svg(svg)

    def handle_delete_monitor(self, monitor_id, token):
        data = load_monitors()
        monitor = None
        for m in data.get("monitors", []):
            if m["id"] == monitor_id:
                monitor = m
                break

        if not monitor:
            self.send_json(404, {"error": "Monitor not found"})
            return

        if monitor.get("delete_token") != token:
            self.send_json(403, {"error": "Invalid delete token"})
            return

        monitor["status"] = "deleted"
        save_monitors(data)

        # Remove status page
        status_path = os.path.join(BASE_DIR, "public", "status", f"{monitor_id}.html")
        if os.path.exists(status_path):
            os.remove(status_path)

        log.info(f"Monitor deleted: {monitor['name']} ({monitor_id})")
        self.send_json(200, {"message": "Monitor deleted", "id": monitor_id})


class ThreadedHTTPServer(HTTPServer):
    """Handle requests in separate threads."""
    def process_request(self, request, client_address):
        thread = threading.Thread(target=self._handle_request, args=(request, client_address))
        thread.daemon = True
        thread.start()

    def _handle_request(self, request, client_address):
        try:
            self.finish_request(request, client_address)
        except Exception:
            self.handle_error(request, client_address)
        finally:
            self.shutdown_request(request)


def main():
    log.info(f"UptimePulse API starting on port {PORT}")
    server = ThreadedHTTPServer(("0.0.0.0", PORT), UptimePulseHandler)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        log.info("Shutting down")
        server.shutdown()


if __name__ == "__main__":
    main()