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

#!/usr/bin/env python3
"""
Demo Manager — FastAPI service for managing demo sites.
Three visibility modes: disabled, internal (192.168.1.x), public.
Pocket ID SSO + API key auth for management.
"""
import ipaddress
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
import _db

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

# ─── Internal network detection ───────────
INTERNAL_NETS = [ipaddress.ip_network("192.168.1.0/24")]


def is_internal_ip(request: Request) -> bool:
    """Check if the request originated from an internal network."""
    forwarded = request.headers.get("X-Forwarded-For", "")
    if forwarded:
        client_ip = forwarded.split(",")[0].strip()
    elif request.client:
        client_ip = request.client.host
    else:
        return False
    try:
        ip = ipaddress.ip_address(client_ip)
        return any(ip in net for net in INTERNAL_NETS)
    except ValueError:
        return False


# ─── Models ─────────────────────────────────────────
class SiteInfo(BaseModel):
    name: str
    visibility: str = "internal"
    description: str = ""
    created_at: str = ""
    updated_at: str = ""


class SiteCreate(BaseModel):
    name: str
    visibility: str = "internal"
    description: str = ""


class VisibilityUpdate(BaseModel):
    visibility: str


# ─── Helpers ────────────────────────────────────────


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:
    info = _db.get(name)
    if not info:
        return False
    site_dir = DEMO_ROOT / name
    try:
        return site_dir.is_dir()
    except OSError:
        return True


def error_resp(status_code: int = 404) -> HTMLResponse:
    resp = HTMLResponse(AUTH_REQUIRED_HTML, status_code=status_code)
    resp.headers["Content-Disposition"] = "inline"
    resp.headers["Cache-Control"] = "no-store"
    return resp


# ─── File Helpers (NFS safe — no rename) ─────────────
def sync_visibility(name: str):
    """Sync the on-disk index.html based on visibility mode."""
    info = _db.get(name)
    if not info:
        return
    v = info["visibility"]
    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 v == "public":
        # Ensure public site has visible index.html
        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:
        # disabled or internal → hide index.html
        if real_index.exists():
            if v == "disabled":
                bak_index.write_bytes(real_index.read_bytes())
                real_index.unlink()
            else:  # internal
                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():
    return HTMLResponse(AUTH_REQUIRED_HTML)


# ─── Site content serving (access control) ──────────

@app.get("/site-content/{path:path}")
async def site_content(path: str = "", request: Request = None):
    """Serve site files with visibility-based access control."""
    if not path:
        raise HTTPException(404)
    parts = path.split("/", 1)
    site_name = parts[0]
    sub = parts[1] if len(parts) > 1 else ""

    info = _db.get(site_name)
    if not info:
        return error_resp(404)

    visibility = info["visibility"]
    user = check_auth(request)

    # ── Access control ──
    if visibility == "disabled":
        # No one can access disabled sites via public URL
        if not user:
            return error_resp(404)
        # Even auth'd users get 404 on disabled sites (use admin UI instead)
        return error_resp(404)

    elif visibility == "internal":
        # Internal IPs can access without auth; external needs auth
        if not user and not is_internal_ip(request):
            # External + no auth → 404 (no login prompt)
            return error_resp(404)

    # visibility == "public" → everyone can access
    # ── Serve file ──
    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():
            return error_resp(404)
    if target.is_dir():
        target = target / "index.html"
        if not target.exists():
            return error_resp(404)
    # HTML files render inline; others (CSS, JS, images) served as files
    if target.suffix == ".html" or target.suffix in (".draft", ".bak"):
        return HTMLResponse(target.read_text())
    resp = FileResponse(target)
    resp.headers["Content-Disposition"] = "inline"
    return resp


# ─── Site management ────────────────────────────────

def _site_to_dict(name: str, info: dict) -> dict:
    """Build the public site dict from DB info + filesystem."""
    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
    return {
        "name": name,
        "visibility": info["visibility"],
        "description": info["description"],
        "created_at": info["created_at"],
        "updated_at": info["updated_at"],
        "exists": True,
        "file_count": file_count,
        "url": f"https://demo.junv.cc/{name}" if info["visibility"] == "public" else None,
        "public": info["visibility"] == "public",
    }


@app.get("/api/sites")
def list_sites(request: Request):
    user = require_auth(request)
    meta = _db.load_all()
    sites = []
    for name in sorted(meta.get("sites", {}).keys()):
        info = meta["sites"][name]
        sites.append(_site_to_dict(name, info))
    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()
    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)
    _db.insert(
        name=name,
        visibility=body.visibility if body.visibility in ("disabled", "internal", "public") else "internal",
        description=body.description,
        now=now,
    )
    sync_visibility(name)
    info = _db.get(name)
    assert info is not None
    return _site_to_dict(name, 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")
    _db.delete(name)
    site_dir = DEMO_ROOT / name
    try:
        shutil.rmtree(site_dir)
    except OSError:
        pass
    return {"status": "deleted", "name": name}


# ── Visibility management ──────────────────────────

@app.post("/api/sites/{name}/visibility")
def set_visibility(name: str, body: VisibilityUpdate, request: Request):
    """Change site visibility: disabled | internal | public."""
    user = require_auth(request)
    if not site_exists(name):
        raise HTTPException(status_code=404)
    v = body.visibility
    if v not in ("disabled", "internal", "public"):
        raise HTTPException(status_code=400, detail="visibility must be: disabled, internal, or public")
    _db.update(name, visibility=v, updated_at=datetime.now(timezone.utc).isoformat())
    sync_visibility(name)
    info = _db.get(name)
    assert info is not None
    return _site_to_dict(name, info)


# ── Backward-compat endpoints (deprecated) ─────────

@app.post("/api/sites/{name}/enable")
def enable_site(name: str, request: Request):
    # Old "enable" → set visibility to last non-disabled value or public
    user = require_auth(request)
    if not site_exists(name):
        raise HTTPException(status_code=404)
    info = _db.get(name)
    v = "public" if info and info.get("published") else "internal"
    _db.update(name, visibility=v, updated_at=datetime.now(timezone.utc).isoformat())
    sync_visibility(name)
    return {"status": "enabled", "name": name, "visibility": v}


@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)
    _db.update(name, visibility="disabled", updated_at=datetime.now(timezone.utc).isoformat())
    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)
    _db.update(name, visibility="public", updated_at=datetime.now(timezone.utc).isoformat())
    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)
    _db.update(name, visibility="internal", updated_at=datetime.now(timezone.utc).isoformat())
    sync_visibility(name)
    return {"status": "unpublished", "name": name}


# ── File management ────────────────────────────────

@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)
    info = _db.get(name)
    if not info:
        raise HTTPException(status_code=404)
    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": info["visibility"] == "public",
        "url": f"https://demo.junv.cc/{name}" if info["visibility"] == "public" 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)
    _db.update(name, updated_at=datetime.now(timezone.utc).isoformat())
    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}


@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")


# ─── Web 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):
    user = check_auth(request)
    if user:
        return RedirectResponse(url="/admin", status_code=302)
    resp = HTMLResponse(AUTH_REQUIRED_HTML, status_code=404)
    resp.headers["Content-Disposition"] = "inline"
    resp.headers["Cache-Control"] = "no-store"
    return resp