ai-agent/backend/projectflags.py

"""Read/write a project's ``feature-flags.json`` — the per-project declaration
of the query-param feature flags its frontend ships behind.

The file lives in the project's **served** directory so the deployed site can
fetch it at ``/feature-flags.json``:

    <project>/public/feature-flags.json     (Vite/CRA — preferred)
    <project>/feature-flags.json            (fallback, plain static sites)

Shape::

    {
      "flags": [
        {
          "key": "archives",              # the query param: ?archives=1
          "label": "Archives",
          "description": "Past-grid browser in the header nav",
          "enabled": false,               # the *for everyone* default
          "since": "2026-08-17"
        }
      ]
    }

``enabled`` is the only field the admin panel flips for everyone; a visitor's
own ``?key=1`` / panel toggle is a client-side override that never touches this
file. Unknown keys on a flag are preserved across a round trip, so a project can
carry extra metadata the editor doesn't know about.
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import os
import re
import secrets
import tempfile
import time
from pathlib import Path
from typing import Any

FILENAME = "feature-flags.json"

# A flag key doubles as a URL query parameter and a storage key, so keep it to
# the characters that are unambiguous in both.
KEY_RE = re.compile(r"^[a-z][a-z0-9-]{0,47}$")

# Fields the editor owns. Anything else on a flag object is passed through.
_KNOWN = ("key", "label", "description", "enabled", "since")


def flags_path(entry: Path) -> Path:
    """Where this project's flags file lives (or would be created).

    An existing file wins wherever it is; otherwise ``public/`` is preferred
    when the project has one (every Vite template does), so a newly declared
    flag is served by the deployed site without a build-config change.
    """
    public = entry / "public" / FILENAME
    root = entry / FILENAME
    if public.is_file():
        return public
    if root.is_file():
        return root
    return public if (entry / "public").is_dir() else root


def _coerce(raw: Any, index: int) -> dict | None:
    """Normalise one entry of the ``flags`` array; None if unusable."""
    if not isinstance(raw, dict):
        return None
    key = raw.get("key")
    if not isinstance(key, str) or not KEY_RE.match(key):
        return None
    flag = dict(raw)
    flag["key"] = key
    flag["label"] = raw["label"] if isinstance(raw.get("label"), str) else key
    flag["description"] = raw["description"] if isinstance(raw.get("description"), str) else ""
    flag["enabled"] = bool(raw.get("enabled"))
    if not isinstance(raw.get("since"), str):
        flag.pop("since", None)
    return flag


def read(entry: Path) -> dict:
    """Parse the project's flags file. A missing file is not an error — it reads
    as an empty, not-yet-created flag set. A malformed one is reported rather
    than silently emptied, so the editor never offers to overwrite a file it
    failed to understand."""
    path = flags_path(entry)
    if not path.is_file():
        return {"path": None, "exists": False, "error": None, "flags": []}
    rel = str(path.relative_to(entry))
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        return {"path": rel, "exists": True, "error": str(exc), "flags": []}
    raw = data.get("flags") if isinstance(data, dict) else data
    if not isinstance(raw, list):
        return {"path": rel, "exists": True, "error": "no `flags` array", "flags": []}
    flags = [f for f in (_coerce(r, i) for i, r in enumerate(raw)) if f is not None]
    # Last write wins on a duplicated key — the file is hand-editable, and a
    # duplicate would otherwise make the editor's save ambiguous.
    seen: dict[str, dict] = {}
    for f in flags:
        seen[f["key"]] = f
    return {"path": rel, "exists": True, "error": None, "flags": list(seen.values())}


def write(entry: Path, flags: list[dict]) -> dict:
    """Replace the project's flag list, preserving any sibling top-level keys
    (``$schema``, project metadata…) already in the file."""
    path = flags_path(entry)
    doc: dict[str, Any] = {}
    if path.is_file():
        try:
            existing = json.loads(path.read_text(encoding="utf-8"))
            if isinstance(existing, dict):
                doc = {k: v for k, v in existing.items() if k != "flags"}
        except (OSError, ValueError):
            doc = {}
    doc["flags"] = flags
    path.parent.mkdir(parents=True, exist_ok=True)
    body = json.dumps(doc, indent=2, ensure_ascii=False) + "\n"
    # Atomic replace so a half-written file can never be served.
    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".flags-", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(body)
        os.replace(tmp, path)
    except BaseException:
        if os.path.exists(tmp):
            os.unlink(tmp)
        raise
    return read(entry)


def merge(current: list[dict], updates: list[dict], allow_create: bool) -> list[dict]:
    """Apply editor updates onto the current list.

    ``updates`` is the full desired list — order included, since the panel shows
    flags in file order. Unknown fields on an existing flag survive because the
    stored object is updated in place rather than rebuilt.
    """
    by_key = {f["key"]: f for f in current}
    out: list[dict] = []
    for upd in updates:
        key = upd.get("key")
        if not isinstance(key, str) or not KEY_RE.match(key):
            raise ValueError(f"invalid flag key: {key!r}")
        prev = by_key.get(key)
        if prev is None and not allow_create:
            raise ValueError(f"unknown flag: {key!r}")
        flag = dict(prev) if prev else {"key": key}
        for field in ("label", "description", "since"):
            if field in upd:
                flag[field] = upd[field]
        if "enabled" in upd:
            flag["enabled"] = bool(upd["enabled"])
        flag.setdefault("label", key)
        flag.setdefault("description", "")
        flag.setdefault("enabled", False)
        out.append(flag)
    return out


# ── admin token ───────────────────────────────────────────────────────────────
# The deployed sites are public and have no session with the lab, so the admin
# panel is unlocked by a token instead: mint it here (behind Authelia), carry it
# to the site once as `?ff-admin=…`, and it lives in that browser's
# localStorage. Showing the panel is a client-side check; *publishing* a flag
# for everyone comes back here and is verified below, which is the boundary that
# actually matters.

TOKEN_PREFIX = "ffa1"
TOKEN_TTL = 365 * 24 * 3600  # a year — this is a bookmarklet-grade credential

_SECRET_PATH = Path(os.environ.get("FLAGS_SECRET_PATH", "/data/flags-secret"))


def _b64(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def _unb64(text: str) -> bytes:
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


def _secret() -> bytes:
    """The HMAC key, created on first use. Deleting the file revokes every token
    ever minted — that is the intended panic button."""
    try:
        return bytes.fromhex(_SECRET_PATH.read_text().strip())
    except (OSError, ValueError):
        pass
    raw = secrets.token_bytes(32)
    _SECRET_PATH.parent.mkdir(parents=True, exist_ok=True)
    # Write via a private temp file so the secret is never briefly world-readable.
    fd, tmp = tempfile.mkstemp(dir=str(_SECRET_PATH.parent), prefix=".secret-")
    try:
        os.chmod(tmp, 0o600)
        with os.fdopen(fd, "w") as fh:
            fh.write(raw.hex())
        os.replace(tmp, _SECRET_PATH)
    except BaseException:
        if os.path.exists(tmp):
            os.unlink(tmp)
        raise
    return raw


def mint_token(subject: str = "gabrielvidal", ttl: int = TOKEN_TTL) -> str:
    now = int(time.time())
    payload = _b64(json.dumps({"sub": subject, "iat": now, "exp": now + ttl}).encode())
    sig = _b64(hmac.new(_secret(), payload.encode(), hashlib.sha256).digest())
    return f"{TOKEN_PREFIX}.{payload}.{sig}"


def verify_token(token: str | None) -> dict | None:
    """Decoded payload for a valid, unexpired token; None otherwise."""
    if not isinstance(token, str):
        return None
    parts = token.split(".")
    if len(parts) != 3 or parts[0] != TOKEN_PREFIX:
        return None
    _, payload, sig = parts
    expected = _b64(hmac.new(_secret(), payload.encode(), hashlib.sha256).digest())
    if not hmac.compare_digest(sig, expected):
        return None
    try:
        data = json.loads(_unb64(payload))
    except (ValueError, TypeError):
        return None
    if not isinstance(data, dict) or int(data.get("exp", 0)) < time.time():
        return None
    return data


def revoke_all() -> None:
    """Rotate the signing key, invalidating every issued token."""
    try:
        _SECRET_PATH.unlink()
    except FileNotFoundError:
        pass


# ── deploy hosts ──────────────────────────────────────────────────────────────
def deploy_hosts(entry: Path) -> list[str]:
    """The hosts this project deploys to, from `package.json` → `zipgo.deploy`.

    These are where an admin link points and where a publish pushes the file.
    """
    try:
        pkg = json.loads((entry / "package.json").read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return []
    deploy = (pkg.get("zipgo") or {}).get("deploy") if isinstance(pkg, dict) else None
    if not isinstance(deploy, dict):
        return []
    return [h for h in deploy if isinstance(h, str) and h]