fix(skills): require admin token and strict URL validation for /api/skills/install

.env.example

@@ -7,3 +7,11 @@ DEFAULT_DEVICE=
 OPENAI_API_KEY=
 ANTHROPIC_API_KEY=
 OPENROUTER_API_KEY=
+
+# Admin token — required by /api/skills/install (and any other endpoint that
+# executes code / installs packages).  Set this to a strong random secret
+# (e.g. `python -c "import secrets; print(secrets.token_urlsafe(32))"`) and
+# send it as the `X-Ghost-Admin-Token` header (or `Authorization: Bearer …`).
+# If unset, /api/skills/install refuses to run — the previous default let any
+# caller on the network drive-by-install a Python package on the host.
+GITD_ADMIN_TOKEN=

gitd/routers/skills.py

@@ -2,6 +2,7 @@
 
 import io
 import json
+import re
 import shutil
 import time
 import zipfile
@@ -13,6 +14,7 @@
 from sqlalchemy.orm import Session
 
 from gitd.models.base import get_db
+from gitd.services.admin_auth import require_admin_token
 
 router = APIRouter(prefix="/api/skills", tags=["skills"])
 
@@ -134,41 +136,86 @@ def api_skills_community():
     return _fetch_cached(_COMMUNITY_URL, "community")
 
 
-@router.post("/install", summary="Install Skill From Registry or URL")
+# Strict shape check for GitHub repo references.  ``_is_github_url`` only
+# checks the prefix, which lets whitespace / newline / leading-dash tricks
+# reach ``git clone`` as smuggled flags (e.g. ``--upload-pack=<cmd>``).
+# Only accept plain ``[https://]github.com/<owner>/<repo>[.git]``.
+_SAFE_GITHUB_URL_RE = re.compile(r"^(?:https?://)?github\.com/[A-Za-z0-9_.\-]+/[A-Za-z0-9_.\-]+(?:\.git)?/?$")
+
+# Registry names are drawn from ``_download_skill_from_registry`` output —
+# keep the accepted alphabet narrow so the value can never do anything
+# surprising when it eventually reaches ``importlib.import_module``.
+_SAFE_REGISTRY_NAME_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_\-]{0,63}$")
+
+
+@router.post(
+    "/install",
+    summary="Install Skill From Registry or URL",
+    dependencies=[Depends(require_admin_token)],
+)
 def api_skills_install(data: dict = Body(...)):
     """Install a skill by name (from registry) or by URL (GitHub repo).
 
     Body: {"name": "tiktok"} or {"url": "github.com/user/repo"}
+
+    Requires the admin token dependency: installing a skill drops Python
+    code into ``gitd/skills/<name>`` which is later loaded via
+    ``importlib.import_module`` — trivially RCE for an unauthenticated
+    caller (CWE-94).  See ``gitd/services/admin_auth.py``.
     """
     from gitd.cli import (
         _clone_github_skill,
         _download_skill_from_registry,
         _install_to_skills_dir,
-        _is_github_url,
+        _validate_skill_dir,
     )
 
-    name = data.get("name", "").strip()
-    url = data.get("url", "").strip()
+    name = data.get("name", "").strip() if isinstance(data.get("name"), str) else ""
+    url = data.get("url", "").strip() if isinstance(data.get("url"), str) else ""
 
     if not name and not url:
         raise HTTPException(status_code=400, detail="Provide 'name' or 'url'")
 
-    if url and _is_github_url(url):
+    if url:
+        # Reject anything that isn't a plain github.com/<owner>/<repo> URL —
+        # ``git clone`` will happily interpret ``-uexec=...`` style tokens
+        # as flags when we haven't sanitised them first.
+        if not _SAFE_GITHUB_URL_RE.match(url):
+            raise HTTPException(
+                status_code=400,
+                detail=("url must look like 'github.com/<owner>/<repo>' (no shell metacharacters, no leading dash)."),
+            )
         source = _clone_github_skill(url)
         if source is None:
             raise HTTPException(status_code=400, detail=f"Failed to clone {url}")
-        ok = _install_to_skills_dir(source)
-        shutil.rmtree(source, ignore_errors=True)
+        try:
+            if not _validate_skill_dir(source, verbose=False):
+                raise HTTPException(
+                    status_code=400,
+                    detail=("Skill failed validation (see server logs); refusing to install untrusted code."),
+                )
+            ok = _install_to_skills_dir(source)
+        finally:
+            shutil.rmtree(source, ignore_errors=True)
         if not ok:
             raise HTTPException(status_code=500, detail="Install failed")
         return {"ok": True, "message": f"Installed from {url}"}
 
     if name:
+        if not _SAFE_REGISTRY_NAME_RE.match(name):
+            raise HTTPException(status_code=400, detail="Invalid registry name")
         source = _download_skill_from_registry(name)
         if source is None:
             raise HTTPException(status_code=404, detail=f"Skill '{name}' not found in registry")
-        ok = _install_to_skills_dir(source, name=name)
-        shutil.rmtree(source, ignore_errors=True)
+        try:
+            if not _validate_skill_dir(source, verbose=False):
+                raise HTTPException(
+                    status_code=400,
+                    detail=("Skill failed validation (see server logs); refusing to install untrusted code."),
+                )
+            ok = _install_to_skills_dir(source, name=name)
+        finally:
+            shutil.rmtree(source, ignore_errors=True)
         if not ok:
             raise HTTPException(status_code=500, detail="Install failed")
         return {"ok": True, "message": f"Installed '{name}' from registry"}

gitd/services/admin_auth.py

@@ -0,0 +1,88 @@
+"""Admin-token authentication for privileged endpoints.
+
+The FastAPI server binds to ``0.0.0.0:5055`` by default and, historically,
+had no authentication at all. That was fine for the "single developer with
+their phone on their desk" use case but became a real problem for endpoints
+that let the caller install code / execute processes on the operator's
+machine — most notably ``POST /api/skills/install`` (CWE-94: attacker URL
+→ ``git clone`` → subsequent ``importlib.import_module`` = RCE).
+
+This module provides a tiny FastAPI ``Depends`` guard that:
+
+  * Reads the shared secret from ``GITD_ADMIN_TOKEN`` at request time
+    (so tests / operators can flip it without restarting the process).
+  * Accepts the token via either the ``X-Ghost-Admin-Token`` header or
+    ``Authorization: Bearer <token>`` (whichever the caller prefers).
+  * Uses ``hmac.compare_digest`` for constant-time comparison.
+  * Refuses the request when the env var is unset — the previous
+    "reachable from anywhere by default" posture must not be silently
+    restored by simply not configuring the token.
+
+Operators who genuinely need the old zero-auth behaviour (e.g. isolated
+CI containers) can opt in with ``GITD_ALLOW_UNAUTHENTICATED_ADMIN=1``.
+It is deliberately noisy so it does not sneak into production.
+"""
+
+from __future__ import annotations
+
+import hmac
+import logging
+import os
+
+from fastapi import Header, HTTPException, status
+
+logger = logging.getLogger(__name__)
+
+_ENV_TOKEN = "GITD_ADMIN_TOKEN"
+_ENV_ALLOW_UNAUTH = "GITD_ALLOW_UNAUTHENTICATED_ADMIN"
+_HEADER = "X-Ghost-Admin-Token"
+
+
+def _extract_bearer(authorization: str | None) -> str | None:
+    """Return the token part of an ``Authorization: Bearer <token>`` header."""
+    if not authorization:
+        return None
+    parts = authorization.strip().split(None, 1)
+    if len(parts) == 2 and parts[0].lower() == "bearer":
+        return parts[1].strip()
+    return None
+
+
+def require_admin_token(
+    x_ghost_admin_token: str | None = Header(default=None, alias=_HEADER),
+    authorization: str | None = Header(default=None),
+) -> None:
+    """FastAPI dependency: enforce a valid admin token.
+
+    Raises ``HTTPException(401)`` when the caller cannot prove possession
+    of the shared secret, or ``HTTPException(500)`` when the operator has
+    left the endpoint unconfigured on a network-reachable server.
+    """
+    if os.environ.get(_ENV_ALLOW_UNAUTH) == "1":
+        # Explicit opt-in for isolated environments only.
+        logger.warning(
+            "admin auth bypassed via %s=1 — do not use on shared hosts",
+            _ENV_ALLOW_UNAUTH,
+        )
+        return
+
+    expected = os.environ.get(_ENV_TOKEN, "").strip()
+    if not expected:
+        # Fail closed. The previous default was "reachable from any origin
+        # the network can route to us", which is what enabled the RCE.
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=(
+                "Admin endpoints are disabled: set the GITD_ADMIN_TOKEN "
+                "environment variable to a strong random secret and send "
+                "it in the X-Ghost-Admin-Token header (or as "
+                "'Authorization: Bearer <token>')."
+            ),
+        )
+
+    supplied = (x_ghost_admin_token or "").strip() or _extract_bearer(authorization) or ""
+    if not supplied or not hmac.compare_digest(supplied, expected):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or missing admin token.",
+        )

tests/api/test_skill_install_auth.py

@@ -0,0 +1,169 @@
+"""Regression tests for CWE-94 in POST /api/skills/install.
+
+The endpoint used to be reachable without any authentication and fed the
+attacker-supplied ``url`` directly into ``git clone``.  Combined with the
+subsequent ``importlib.import_module`` in ``_load_skill`` that was a
+drive-by RCE.  These tests lock in the fix:
+
+* the endpoint refuses to run when ``GITD_ADMIN_TOKEN`` is unset,
+* it refuses invalid tokens,
+* it rejects URLs that don't match a strict shape *before* invoking git
+  (so ``--upload-pack=...``, whitespace-smuggled shell payloads, and
+  hostnames like ``github.com.evil.example.com`` never reach the sink),
+* the intended admin flow still works with a valid token via either
+  ``X-Ghost-Admin-Token`` or ``Authorization: Bearer <token>``.
+"""
+
+from __future__ import annotations
+
+import subprocess
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    """Fresh TestClient with a clean env for each case."""
+    monkeypatch.delenv("GITD_ADMIN_TOKEN", raising=False)
+    monkeypatch.delenv("GITD_ALLOW_UNAUTHENTICATED_ADMIN", raising=False)
+
+    from gitd.app import app  # imported after env is scrubbed
+
+    with TestClient(app) as c:
+        yield c
+
+
+@pytest.fixture()
+def no_git(monkeypatch):
+    """Explode if the router ever lets ``git clone`` execute during a test."""
+    calls: list[list[str]] = []
+    real_run = subprocess.run
+
+    def guarded_run(cmd, *args, **kwargs):
+        if isinstance(cmd, (list, tuple)) and cmd and cmd[0] == "git":
+            calls.append(list(cmd))
+            raise AssertionError(f"git reached with attacker input: {cmd!r}")
+        return real_run(cmd, *args, **kwargs)
+
+    monkeypatch.setattr(subprocess, "run", guarded_run)
+    return calls
+
+
+# ── The drive-by CVE cases ────────────────────────────────────────────
+
+
+def test_unauthenticated_url_install_is_rejected(client, no_git):
+    """No token configured → refuse and never call git."""
+    r = client.post(
+        "/api/skills/install",
+        json={"url": "github.com/attacker/malicious-skill"},
+    )
+    assert r.status_code in (401, 403), r.text
+    assert no_git == []
+
+
+def test_unauthenticated_registry_install_is_rejected(client, no_git):
+    r = client.post(
+        "/api/skills/install",
+        json={"name": "some_registry_skill"},
+    )
+    assert r.status_code in (401, 403), r.text
+
+
+def test_wrong_token_is_rejected(client, no_git, monkeypatch):
+    monkeypatch.setenv("GITD_ADMIN_TOKEN", "s3cret")
+    r = client.post(
+        "/api/skills/install",
+        headers={"X-Ghost-Admin-Token": "not-the-token"},
+        json={"url": "github.com/attacker/malicious-skill"},
+    )
+    assert r.status_code in (401, 403), r.text
+    assert no_git == []
+
+
+def test_flag_injection_url_is_rejected(client, no_git, monkeypatch):
+    """``--upload-pack=<cmd>`` smuggling must be caught before git runs."""
+    monkeypatch.setenv("GITD_ADMIN_TOKEN", "s3cret")
+    r = client.post(
+        "/api/skills/install",
+        headers={"X-Ghost-Admin-Token": "s3cret"},
+        json={"url": "--upload-pack=touch /tmp/pwned github.com/x/y"},
+    )
+    assert r.status_code == 400, r.text
+    assert no_git == []
+
+
+def test_whitespace_smuggling_url_is_rejected(client, no_git, monkeypatch):
+    monkeypatch.setenv("GITD_ADMIN_TOKEN", "s3cret")
+    r = client.post(
+        "/api/skills/install",
+        headers={"X-Ghost-Admin-Token": "s3cret"},
+        json={"url": "github.com/foo/bar\nrm -rf /"},
+    )
+    assert r.status_code == 400, r.text
+    assert no_git == []
+
+
+def test_lookalike_hostname_is_rejected(client, no_git, monkeypatch):
+    """``github.com.evil.example.com`` used to pass the substring check."""
+    monkeypatch.setenv("GITD_ADMIN_TOKEN", "s3cret")
+    r = client.post(
+        "/api/skills/install",
+        headers={"X-Ghost-Admin-Token": "s3cret"},
+        json={"url": "https://github.com.evil.example.com/foo/bar"},
+    )
+    # Either rejected as bad shape (400) or as "not a github URL"; the
+    # important thing is that no clone ran.
+    assert r.status_code in (400, 401), r.text
+    assert no_git == []
+
+
+# ── The legitimate admin flow still works ─────────────────────────────
+
+
+def test_valid_token_via_header_reaches_install(client, monkeypatch, tmp_path):
+    monkeypatch.setenv("GITD_ADMIN_TOKEN", "s3cret")
+
+    from gitd import cli as cli_module
+
+    fake_source = tmp_path / "fake_skill"
+    fake_source.mkdir()
+    (fake_source / "skill.yaml").write_text(
+        "name: fake\ndisplay_name: fake\ndescription: x\nversion: 1.0.0\napp_package: com.example\nauthor: tester\n"
+    )
+
+    monkeypatch.setattr(cli_module, "_clone_github_skill", lambda u: fake_source)
+    monkeypatch.setattr(cli_module, "_install_to_skills_dir", lambda s, name=None: True)
+    monkeypatch.setattr(cli_module, "_validate_skill_dir", lambda s, verbose=True: True)
+
+    r = client.post(
+        "/api/skills/install",
+        headers={"X-Ghost-Admin-Token": "s3cret"},
+        json={"url": "github.com/legit/skill"},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json().get("ok") is True
+
+
+def test_valid_token_via_bearer_header_reaches_install(client, monkeypatch, tmp_path):
+    monkeypatch.setenv("GITD_ADMIN_TOKEN", "s3cret")
+
+    from gitd import cli as cli_module
+
+    fake_source = tmp_path / "fake_skill"
+    fake_source.mkdir()
+    (fake_source / "skill.yaml").write_text(
+        "name: fake\ndisplay_name: fake\ndescription: x\nversion: 1.0.0\napp_package: com.example\nauthor: tester\n"
+    )
+
+    monkeypatch.setattr(cli_module, "_clone_github_skill", lambda u: fake_source)
+    monkeypatch.setattr(cli_module, "_install_to_skills_dir", lambda s, name=None: True)
+    monkeypatch.setattr(cli_module, "_validate_skill_dir", lambda s, verbose=True: True)
+
+    r = client.post(
+        "/api/skills/install",
+        headers={"Authorization": "Bearer s3cret"},
+        json={"url": "github.com/legit/skill"},
+    )
+    assert r.status_code == 200, r.text