feat(webui): auth and deployment boundary (#435)

Add bind-host assessment that refuses 0.0.0.0/:: without override,
warns on non-loopback binds, and documents internal-only MVP serving.
Health endpoint exposes deployment metadata; docs cover Access/VPN/WARP
and runtime env assumptions without embedding secrets in the client.

Closes #435
This commit is contained in:
2026-07-09 11:50:40 -04:00
parent 8d1098f916
commit 975674d7f5
7 changed files with 362 additions and 3 deletions
+14
View File
@@ -2,16 +2,30 @@
from __future__ import annotations
import logging
import os
import uvicorn
from webui.app import create_app
from webui.deployment_boundary import assess_bind_host
logger = logging.getLogger("webui")
def _validate_bind_host(host: str) -> None:
assessment = assess_bind_host(host)
if assessment.disposition == "refuse":
logger.error("webui bind refused: %s", assessment.message)
raise SystemExit(1)
if assessment.disposition == "warn":
logger.warning("webui bind warning: %s", assessment.message)
def main() -> None:
host = os.environ.get("WEBUI_HOST", "127.0.0.1")
port = int(os.environ.get("WEBUI_PORT", "8765"))
_validate_bind_host(host)
uvicorn.run(create_app(), host=host, port=port, log_level="info")
+11 -3
View File
@@ -9,6 +9,7 @@ from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, Response
from starlette.routing import Route
from webui.deployment_boundary import deployment_snapshot
from webui.layout import render_page
from webui.project_registry import find_project, load_registry, registry_to_dict
from webui.project_views import render_project_detail, render_projects_list
@@ -61,11 +62,13 @@ async def home(_request: Request) -> HTMLResponse:
async def health(_request: Request) -> JSONResponse:
bind_host = _request.app.state.webui_bind_host
return JSONResponse({
"status": "ok",
"service": "mcp-control-plane-webui",
"mode": "read-only-mvp",
"timestamp": datetime.now(timezone.utc).isoformat(),
"deployment": deployment_snapshot(bind_host=bind_host),
})
@@ -250,9 +253,12 @@ async def method_not_allowed(request: Request, _exc: Exception) -> Response:
return JSONResponse({"error": "not_found"}, status_code=404)
def create_app() -> Starlette:
def create_app(*, bind_host: str | None = None) -> Starlette:
"""Build the read-only MVP Starlette app."""
return Starlette(
import os
resolved_bind = bind_host or os.environ.get("WEBUI_HOST", "127.0.0.1")
app = Starlette(
debug=False,
routes=[
Route("/", home, methods=["GET"]),
@@ -287,4 +293,6 @@ def create_app() -> Starlette:
Route("/api/leases", api_leases, methods=["GET"]),
],
exception_handlers={405: method_not_allowed},
)
)
app.state.webui_bind_host = resolved_bind
return app
+155
View File
@@ -0,0 +1,155 @@
"""Internal-only deployment boundary for the operator web UI (#435)."""
from __future__ import annotations
import ipaddress
import os
import re
from dataclasses import asdict, dataclass
from typing import Literal
Disposition = Literal["safe", "warn", "refuse"]
_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})
_ALL_INTERFACE_HOSTS = frozenset({"0.0.0.0", "::", "*"})
_ALLOW_PUBLIC_BIND_ENV = "WEBUI_ALLOW_PUBLIC_BIND"
_ALLOW_REMOTE_BIND_ENV = "WEBUI_ALLOW_REMOTE_BIND"
_FORBIDDEN_CLIENT_PATTERNS = (
re.compile(r"GITEA_(?:TOKEN|PASS|PASSWORD)", re.I),
re.compile(r"Bearer\s+[A-Za-z0-9._\-]{20,}"),
re.compile(r"password\s*[:=]\s*['\"][^'\"]+['\"]", re.I),
)
@dataclass(frozen=True)
class BindAssessment:
host: str
disposition: Disposition
message: str
override_env: str | None = None
def _truthy_env(name: str) -> bool:
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes"}
def assess_bind_host(host: str) -> BindAssessment:
"""Classify a bind host as safe, warn, or refuse (fail closed on 0.0.0.0/::)."""
normalized = (host or "").strip().lower()
if not normalized:
return BindAssessment(
host=host,
disposition="refuse",
message="WEBUI_HOST must not be empty.",
)
if normalized in _LOOPBACK_HOSTS:
return BindAssessment(
host=host,
disposition="safe",
message="Loopback bind — suitable for local operator console.",
)
if normalized in _ALL_INTERFACE_HOSTS:
if _truthy_env(_ALLOW_PUBLIC_BIND_ENV):
return BindAssessment(
host=host,
disposition="warn",
message=(
"Binding all interfaces. Protect with Cloudflare Access, "
"WARP, VPN, or equivalent before exposing beyond localhost."
),
override_env=_ALLOW_PUBLIC_BIND_ENV,
)
return BindAssessment(
host=host,
disposition="refuse",
message=(
f"Refusing all-interface bind ({host!r}). Set "
f"{_ALLOW_PUBLIC_BIND_ENV}=1 only when fronted by trusted "
"network access controls."
),
override_env=_ALLOW_PUBLIC_BIND_ENV,
)
try:
addr = ipaddress.ip_address(normalized)
if addr.is_loopback:
return BindAssessment(
host=host,
disposition="safe",
message="Loopback bind — suitable for local operator console.",
)
except ValueError:
pass
if _truthy_env(_ALLOW_REMOTE_BIND_ENV):
return BindAssessment(
host=host,
disposition="warn",
message=(
"Non-loopback bind acknowledged. Restrict to a trusted network "
"and add Cloudflare Access, WARP, or VPN if reachable beyond it."
),
override_env=_ALLOW_REMOTE_BIND_ENV,
)
return BindAssessment(
host=host,
disposition="warn",
message=(
f"Non-loopback bind ({host!r}). MVP expects 127.0.0.1; set "
f"{_ALLOW_REMOTE_BIND_ENV}=1 to acknowledge a trusted-network bind."
),
override_env=_ALLOW_REMOTE_BIND_ENV,
)
def runtime_assumptions() -> dict[str, str]:
"""Documented runtime paths and hosts — never includes secrets."""
repo_root = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
registry = (os.environ.get("WEBUI_PROJECT_REGISTRY") or "").strip()
profile_config = (os.environ.get("GITEA_MCP_CONFIG") or "").strip()
profile_name = (os.environ.get("GITEA_MCP_PROFILE") or "").strip()
return {
"deployment_mode": "internal-operator-console",
"webui_host_env": "WEBUI_HOST",
"webui_port_env": "WEBUI_PORT",
"webui_repo_root_env": "WEBUI_REPO_ROOT",
"webui_repo_root": repo_root or "(defaults to repository root)",
"webui_project_registry_env": "WEBUI_PROJECT_REGISTRY",
"webui_project_registry": registry or "(packaged webui/data/projects.registry.json)",
"gitea_mcp_config_env": "GITEA_MCP_CONFIG",
"gitea_mcp_config": profile_config or "(optional; server-side only)",
"gitea_mcp_profile_env": "GITEA_MCP_PROFILE",
"gitea_mcp_profile": profile_name or "(optional; server-side only)",
"gitea_credentials": "Resolved server-side via gitea_auth; never embedded in HTML/JS",
"public_bind_override_env": _ALLOW_PUBLIC_BIND_ENV,
"remote_bind_override_env": _ALLOW_REMOTE_BIND_ENV,
}
def scan_text_for_client_secrets(text: str) -> list[str]:
"""Return human-readable findings if *text* looks like it embeds secrets."""
findings: list[str] = []
for pattern in _FORBIDDEN_CLIENT_PATTERNS:
if pattern.search(text):
findings.append(f"matched forbidden client pattern: {pattern.pattern}")
return findings
def deployment_snapshot(*, bind_host: str | None = None) -> dict[str, object]:
host = bind_host if bind_host is not None else os.environ.get("WEBUI_HOST", "127.0.0.1")
bind = assess_bind_host(host)
return {
"mode": "internal-operator-console",
"mvp_auth": "none",
"bind": asdict(bind),
"runtime_assumptions": runtime_assumptions(),
"client_secret_policy": (
"No Gitea tokens, passwords, or profile secrets in HTML, JS, "
"or browser storage."
),
}