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"),
    # Router's WAN IP — internal traffic via NAT appears from this IP
    ipaddress.ip_network("14.137.198.99/32"),
]


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()
    else:
        # Fallback: X-Real-IP (set by nginx ingress) or direct connection
        client_ip = request.headers.get("X-Real-IP", "")
    if not client_ip:
        if 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


# ── Static file fallback ─────────────────────────────
def _find_static_file(path: str):
    """Try to locate a static file (image, css, js, etc.) referenced
    from a site's HTML as a root-relative URL (e.g. /image.jpg from page /site-name).
    Searches all site directories for a matching filename.
    Returns (site_name, sub_path) or None."""
    filename = path.rsplit("/", 1)[-1]
    if not filename or "." not in filename:
        return None
    ext = filename.rsplit(".", 1)[-1].lower()
    if ext not in ("jpg", "jpeg", "png", "gif", "webp", "css", "js", "svg", "ico", "woff2", "pdf"):
        return None
    meta = _db.load_all()
    for name in sorted(meta.get("sites", {}).keys()):
        site_dir = DEMO_ROOT / name
        if not site_dir.is_dir():
            continue
        candidate = site_dir / filename
        if candidate.exists() and candidate.is_file():
            return (name, filename)
    return None


# ─── 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:
        # ── Fallback: if the path doesn't match a site name, it might be
        #    a static file (image, css, js) referenced from a site's HTML.
        #    Try to find it in any site directory.
        found = _find_static_file(path)
        if found:
            site_name, sub = found
            info = _db.get(site_name)
            if not info:
                return error_resp(404)
        else:
            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"):
        resp = HTMLResponse(target.read_text())
        resp.headers["Cache-Control"] = "no-store"
        return resp
    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