home-docker/home-apps/demo-service/service.py

#!/usr/bin/env python3
"""
Demo Manager — FastAPI service for managing demo sites.
Sites start as drafts (unpublished). Publishing makes them public.
Pocket ID SSO + API key auth.
"""
import json
import os
import shutil
import httpx
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, Request, HTTPException, UploadFile, File, Form
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, FileResponse
from pydantic import BaseModel

DEMO_ROOT = Path(os.environ.get("DEMO_ROOT", "/data/demos"))
META_FILE = DEMO_ROOT / ".meta" / "sites.json"
# Config from separate file (avoids secret redaction issues)
from _config import API_KEY, OAUTH_PROXY
from _loaders import AUTH_REQUIRED_HTML, UI_HTML

app = FastAPI(title="Demo Manager", version="1.2.0")

# ─── Models ─────────────────────────────────────────
class SiteInfo(BaseModel):
    name: str
    enabled: bool = True
    published: bool = False
    description: str = ""
    created_at: str = ""
    updated_at: str = ""

class SiteCreate(BaseModel):
    name: str
    description: str = ""
    published: bool = False

# ─── Helpers ────────────────────────────────────────
def load_meta() -> dict:
    META_FILE.parent.mkdir(parents=True, exist_ok=True)
    if META_FILE.exists():
        return json.loads(META_FILE.read_text())
    return {"sites": {}}

def save_meta(meta: dict):
    META_FILE.parent.mkdir(parents=True, exist_ok=True)
    META_FILE.write_text(json.dumps(meta, indent=2))

def check_auth(request: Request) -> Optional[str]:
    auth_header = request.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        token = auth_header[7:]
        if API_KEY and token == API_KEY:
            return "api-agent"
    email = request.headers.get("X-Auth-Request-Email") or request.headers.get("X-Forwarded-Email")
    if email:
        return email
    oauth_cookie = request.cookies.get("_oauth2_proxy")
    if oauth_cookie:
        try:
            resp = httpx.get(
                f"{OAUTH_PROXY}/oauth2/auth",
                cookies={"_oauth2_proxy": oauth_cookie},
                timeout=5,
            )
            if resp.status_code == 202:
                user_email = resp.headers.get("X-Auth-Request-Email", "")
                if user_email:
                    return user_email
        except Exception:
            pass
    return None

def require_auth(request: Request) -> str:
    user = check_auth(request)
    if not user:
        raise HTTPException(status_code=401, detail="Authentication required")
    return user

def site_exists(name: str) -> bool:
    meta = load_meta()
    if name not in meta.get("sites", {}):
        return False
    site_dir = DEMO_ROOT / name
    try:
        return site_dir.is_dir()
    except OSError:
        return True

def is_site_visible(name: str) -> bool:
    meta = load_meta()
    info = meta.get("sites", {}).get(name, {})
    return info.get("enabled", True) and info.get("published", False)

# ─── File Helpers (NFS safe — no rename) ─────────────
def sync_visibility(name: str):
    meta = load_meta()
    info = meta.get("sites", {}).get(name, {})
    visible = info.get("enabled", True) and info.get("published", False)
    site_dir = DEMO_ROOT / name
    real_index = site_dir / "index.html"
    draft_index = site_dir / ".index.html.draft"
    bak_index = site_dir / ".index.html.bak"

    if visible:
        for src in [draft_index, bak_index]:
            if src.exists() and not real_index.exists():
                real_index.write_bytes(src.read_bytes())
                src.unlink()
                break
    else:
        if real_index.exists():
            if not info.get("enabled", True):
                bak_index.write_bytes(real_index.read_bytes())
                real_index.unlink()
            elif not info.get("published", False):
                draft_index.write_bytes(real_index.read_bytes())
                real_index.unlink()

    for leftover in [draft_index, bak_index]:
        if leftover.exists() and real_index.exists():
            leftover.unlink()

# ─── API Routes ─────────────────────────────────────
@app.get("/api/health")
def health():
    return {"status": "ok", "root": str(DEMO_ROOT)}

@app.get("/api/error-page", response_class=HTMLResponse)
def error_page():
    """Serve the canvas error page for any HTTP error."""
    return HTMLResponse(AUTH_REQUIRED_HTML)

@app.get("/site-content", response_class=FileResponse)
@app.get("/site-content/{path:path}")
async def site_content(path: str = "", request: Request = None):
    """Serve site files. Drafts require auth."""
    if not path:
        raise HTTPException(404)
    parts = path.split("/", 1)
    site_name = parts[0]
    sub = parts[1] if len(parts) > 1 else ""
    if not site_name or not site_exists(site_name):
        resp = HTMLResponse(AUTH_REQUIRED_HTML, status_code=404)
        resp.headers["Content-Disposition"] = "inline"
        resp.headers["Cache-Control"] = "no-store"
        return resp
    visible = is_site_visible(site_name)
    if not visible:
        if not check_auth(request):
            resp = HTMLResponse(AUTH_REQUIRED_HTML, status_code=401)
            resp.headers["Content-Disposition"] = "inline"
            resp.headers["Cache-Control"] = "no-store"
            return resp
    site_dir = DEMO_ROOT / site_name
    target = (site_dir / sub) if sub else (site_dir / "index.html")
    if not target.exists():
        if not sub or sub == "index.html":
            for fb in [".index.html.draft", ".index.html.bak"]:
                p = site_dir / fb
                if p.exists():
                    target = p
                    break
        if not target.exists():
            resp = HTMLResponse(AUTH_REQUIRED_HTML, status_code=404)
            resp.headers["Content-Disposition"] = "inline"
            resp.headers["Cache-Control"] = "no-store"
            return resp
    if target.is_dir():
        target = target / "index.html"
        if not target.exists():
            resp = HTMLResponse(AUTH_REQUIRED_HTML, status_code=404)
            resp.headers["Content-Disposition"] = "inline"
            resp.headers["Cache-Control"] = "no-store"
            return resp
    # HTML files render inline; others (CSS, JS, images) are served as files
    if target.suffix == ".html":
        return HTMLResponse(target.read_text())
    resp = FileResponse(target)
    resp.headers["Content-Disposition"] = "inline"
    return resp

@app.get("/api/sites")
def list_sites(request: Request):
    user = require_auth(request)
    meta = load_meta()
    sites = []
    for name in sorted(meta.get("sites", {}).keys()):
        info = meta["sites"][name]
        site_dir = DEMO_ROOT / name
        file_count = 0
        if site_dir.is_dir():
            try:
                file_count = sum(1 for _ in site_dir.rglob("*") if _.is_file() and not _.name.startswith("."))
            except OSError:
                pass
        if "published" not in info:
            info["published"] = True
        sites.append({
            **info, "name": name,
            "exists": True,
            "file_count": file_count,
            "url": f"https://demo.junv.cc/{name}" if is_site_visible(name) else None,
            "public": is_site_visible(name),
        })
    return {"sites": sites, "user": user}

@app.post("/api/sites")
def create_site(body: SiteCreate, request: Request):
    user = require_auth(request)
    name = body.name.strip().lower().replace(" ", "-")
    if not name:
        raise HTTPException(status_code=400, detail="Name is required")
    if site_exists(name):
        raise HTTPException(status_code=409, detail=f"Site '{name}' already exists")
    now = datetime.now(timezone.utc).isoformat()
    meta = load_meta()
    meta.setdefault("sites", {})
    site_dir = DEMO_ROOT / name
    site_dir.mkdir(parents=True, exist_ok=True)
    default_html = f"""<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>{name}</title>
<style>body{{font-family:sans-serif;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0;background:#f5f5f5}}h1{{color:#333}}</style>
</head>
<body><h1>🚀 {name}</h1></body>
</html>"""
    (site_dir / "index.html").write_text(default_html)
    site_info = {
        "name": name, "enabled": True, "published": body.published,
        "description": body.description, "created_at": now, "updated_at": now,
    }
    meta["sites"][name] = site_info
    save_meta(meta)
    sync_visibility(name)
    return site_info

@app.delete("/api/sites/{name}")
def delete_site(name: str, request: Request):
    user = require_auth(request)
    if not site_exists(name):
        raise HTTPException(status_code=404, detail=f"Site '{name}' not found")
    meta = load_meta()
    meta["sites"].pop(name, None)
    save_meta(meta)
    site_dir = DEMO_ROOT / name
    try: shutil.rmtree(site_dir)
    except OSError: pass
    return {"status": "deleted", "name": name}

@app.post("/api/sites/{name}/enable")
def enable_site(name: str, request: Request):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    meta = load_meta()
    if name in meta.get("sites", {}):
        meta["sites"][name]["enabled"] = True
        meta["sites"][name]["updated_at"] = datetime.now(timezone.utc).isoformat()
        save_meta(meta)
    sync_visibility(name)
    return {"status": "enabled", "name": name}

@app.post("/api/sites/{name}/disable")
def disable_site(name: str, request: Request):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    meta = load_meta()
    if name in meta.get("sites", {}):
        meta["sites"][name]["enabled"] = False
        meta["sites"][name]["updated_at"] = datetime.now(timezone.utc).isoformat()
        save_meta(meta)
    sync_visibility(name)
    return {"status": "disabled", "name": name}

@app.post("/api/sites/{name}/publish")
def publish_site(name: str, request: Request):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    meta = load_meta()
    if name in meta.get("sites", {}):
        meta["sites"][name]["published"] = True
        meta["sites"][name]["updated_at"] = datetime.now(timezone.utc).isoformat()
        save_meta(meta)
    sync_visibility(name)
    return {"status": "published", "name": name, "url": f"https://demo.junv.cc/{name}"}

@app.post("/api/sites/{name}/unpublish")
def unpublish_site(name: str, request: Request):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    meta = load_meta()
    if name in meta.get("sites", {}):
        meta["sites"][name]["published"] = False
        meta["sites"][name]["updated_at"] = datetime.now(timezone.utc).isoformat()
        save_meta(meta)
    sync_visibility(name)
    return {"status": "unpublished", "name": name}

@app.get("/api/sites/{name}/preview")
def preview_site(name: str, request: Request):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    site_dir = DEMO_ROOT / name
    for candidate in ["index.html", ".index.html.draft", ".index.html.bak"]:
        fpath = site_dir / candidate
        if fpath.exists():
            return HTMLResponse(fpath.read_text())
    raise HTTPException(status_code=404, detail="No content")

@app.get("/api/sites/{name}")
def get_site(name: str, request: Request):
    require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    meta = load_meta()
    info = meta.get("sites", {}).get(name, {"name": name, "enabled": True, "published": False})
    if "published" not in info:
        info["published"] = True
    site_dir = DEMO_ROOT / name
    files = []
    if site_dir.is_dir():
        try:
            for f in sorted(site_dir.rglob("*")):
                if f.is_file() and not f.name.startswith("."):
                    rel = str(f.relative_to(site_dir))
                    files.append({"path": rel, "size": f.stat().st_size})
        except OSError: pass
    return {
        "site": info, "files": files,
        "public": is_site_visible(name),
        "url": f"https://demo.junv.cc/{name}" if is_site_visible(name) else None,
    }

@app.post("/api/sites/{name}/files")
async def upload_file(name: str, request: Request, file: UploadFile = File(...), path: str = Form("")):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404, detail=f"Site '{name}' not found")
    site_dir = DEMO_ROOT / name
    safe_path = path.strip("/").replace("..", "")
    dest_dir = site_dir / safe_path
    dest_dir.mkdir(parents=True, exist_ok=True)
    file_path = dest_dir / file.filename
    content = await file.read()
    file_path.write_bytes(content)
    meta = load_meta()
    if name in meta.get("sites", {}):
        meta["sites"][name]["updated_at"] = datetime.now(timezone.utc).isoformat()
        save_meta(meta)
    return {"status": "uploaded", "name": name, "file": str(file_path.relative_to(site_dir)), "size": len(content)}

@app.delete("/api/sites/{name}/files")
def delete_file(name: str, request: Request, path: str = Form("")):
    user = require_auth(request)
    if not site_exists(name): raise HTTPException(status_code=404)
    safe_path = path.strip("/").replace("..", "")
    if not safe_path: raise HTTPException(status_code=400, detail="path required")
    file_path = DEMO_ROOT / name / safe_path
    if not file_path.exists(): raise HTTPException(status_code=404)
    if file_path.is_dir(): shutil.rmtree(file_path)
    else: file_path.unlink()
    return {"status": "deleted", "name": name, "file": safe_path}

# ─── Routes ─────────────────────────────────────────

@app.get("/admin", response_class=HTMLResponse)
@app.get("/admin/", response_class=HTMLResponse)
@app.get("/admin/{path:path}", response_class=HTMLResponse)
def admin_ui(request: Request, path: str = ""):
    if not check_auth(request):
        return HTMLResponse(AUTH_REQUIRED_HTML)
    return HTMLResponse(UI_HTML)

@app.get("/")
def index(request: Request):
    return HTMLResponse("""<!DOCTYPE html>
<html><head><meta http-equiv="refresh" content="0;url=/admin"></head>
<body><p>Redirecting to <a href="/admin">Admin</a>...</p></body>
</html>""")