Compare commits
5
Commits
fe29379ad2
...
64b83cd1fb
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
64b83cd1fb | ||
|
|
23535e30bc | ||
|
|
351ef3899b | ||
|
|
1334b0b320 | ||
|
|
975674d7f5 |
@@ -54,6 +54,25 @@ Use `-q` for a compact summary and `-v` to see individual test names.
|
|||||||
./venv/bin/python -m pytest tests/ -q
|
./venv/bin/python -m pytest tests/ -q
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### Web UI suite (#436)
|
||||||
|
|
||||||
|
Hermetic unittest modules matching `test_webui_*.py` cover route rendering,
|
||||||
|
read-only guards, registry/prompt/queue loaders, and optional child-issue
|
||||||
|
modules when present on the branch.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
./scripts/test-webui
|
||||||
|
./scripts/ci-webui-check # skip unless the diff touches web UI paths
|
||||||
|
WEBUI_CI_FORCE=1 ./scripts/ci-webui-check
|
||||||
|
```
|
||||||
|
|
||||||
|
`scripts/test-webui` defaults `WEBUI_TEST_OFFLINE=1` so route coverage never
|
||||||
|
needs Gitea credentials or MCP daemon credential access. Set
|
||||||
|
`WEBUI_TEST_OFFLINE=0` only for explicit operator live-fetch checks.
|
||||||
|
|
||||||
|
Wire `scripts/ci-webui-check` into Jenkins (or equivalent) for PRs that touch
|
||||||
|
`webui/`, `tests/test_webui_*`, or `docs/webui*`.
|
||||||
|
|
||||||
### Run targeted tests
|
### Run targeted tests
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -0,0 +1,59 @@
|
|||||||
|
# Web UI deployment boundary (#435)
|
||||||
|
|
||||||
|
The MCP Control Plane web UI is an **internal operator console**, not a
|
||||||
|
customer-facing application. The MVP assumes local or trusted-network access
|
||||||
|
only.
|
||||||
|
|
||||||
|
## MVP deployment model
|
||||||
|
|
||||||
|
- **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`)
|
||||||
|
- **Authentication:** none in MVP — protection comes from network placement
|
||||||
|
- **Mutations:** read-only routes; gated write actions remain disabled (#434)
|
||||||
|
- **Secrets:** resolved server-side via `gitea_auth` / `GITEA_MCP_CONFIG`; never
|
||||||
|
embedded in HTML, JavaScript, or browser storage
|
||||||
|
|
||||||
|
Do **not** expose the UI on the public internet without an access layer.
|
||||||
|
|
||||||
|
## Beyond localhost
|
||||||
|
|
||||||
|
If the UI must be reachable outside the operator laptop:
|
||||||
|
|
||||||
|
1. Prefer **Cloudflare Access**, **Cloudflare WARP**, or an org **VPN** so only
|
||||||
|
authenticated staff reach the service.
|
||||||
|
2. Bind to a specific interface only when necessary — never `0.0.0.0` / `::`
|
||||||
|
without understanding the exposure.
|
||||||
|
3. Set explicit override env vars only after access controls are in place:
|
||||||
|
- `WEBUI_ALLOW_PUBLIC_BIND=1` — acknowledges all-interface bind (`0.0.0.0`, `::`)
|
||||||
|
- `WEBUI_ALLOW_REMOTE_BIND=1` — acknowledges a non-loopback host
|
||||||
|
|
||||||
|
Startup **refuses** all-interface binds unless `WEBUI_ALLOW_PUBLIC_BIND=1`.
|
||||||
|
Non-loopback binds log a warning unless `WEBUI_ALLOW_REMOTE_BIND=1`.
|
||||||
|
|
||||||
|
## Environment variables
|
||||||
|
|
||||||
|
| Variable | Default | Purpose |
|
||||||
|
|----------|---------|---------|
|
||||||
|
| `WEBUI_HOST` | `127.0.0.1` | Bind address |
|
||||||
|
| `WEBUI_PORT` | `8765` | Listen port |
|
||||||
|
| `WEBUI_REPO_ROOT` | repository root | Workflow/schema hash root for prompt library |
|
||||||
|
| `WEBUI_PROJECT_REGISTRY` | packaged JSON | Project registry file path |
|
||||||
|
| `GITEA_MCP_CONFIG` | unset | Server-side MCP profile config path (optional) |
|
||||||
|
| `GITEA_MCP_PROFILE` | unset | Active MCP profile name (optional) |
|
||||||
|
| `WEBUI_ALLOW_PUBLIC_BIND` | unset | Acknowledge `0.0.0.0` / `::` bind |
|
||||||
|
| `WEBUI_ALLOW_REMOTE_BIND` | unset | Acknowledge non-loopback bind |
|
||||||
|
|
||||||
|
Gitea credentials (`GITEA_TOKEN_*`, `.env.*`, keychain refs) are read only on
|
||||||
|
the server when a page needs live Gitea data (e.g. `/queue`). They are not
|
||||||
|
shipped to the browser.
|
||||||
|
|
||||||
|
## Health / deployment metadata
|
||||||
|
|
||||||
|
`GET /health` includes a `deployment` object with bind disposition, runtime
|
||||||
|
assumption paths, and the client-secret policy. Use it to verify an instance is
|
||||||
|
configured for internal-only operation.
|
||||||
|
|
||||||
|
## Non-goals (MVP)
|
||||||
|
|
||||||
|
- Full SSO or session login in the UI
|
||||||
|
- Hosting on the public internet without Access/VPN/WARP
|
||||||
|
- Embedding Gitea tokens in the frontend bundle
|
||||||
+66
-1
@@ -29,6 +29,13 @@ Optional environment variables:
|
|||||||
|----------|---------|---------|
|
|----------|---------|---------|
|
||||||
| `WEBUI_HOST` | `127.0.0.1` | Bind address (keep local for MVP) |
|
| `WEBUI_HOST` | `127.0.0.1` | Bind address (keep local for MVP) |
|
||||||
| `WEBUI_PORT` | `8765` | Listen port |
|
| `WEBUI_PORT` | `8765` | Listen port |
|
||||||
|
| `WEBUI_REPO_ROOT` | repository root | Prompt library workflow hash root |
|
||||||
|
| `WEBUI_PROJECT_REGISTRY` | packaged JSON | Project registry path |
|
||||||
|
| `GITEA_MCP_CONFIG` | unset | Server-side MCP profile config (never sent to browser) |
|
||||||
|
| `GITEA_MCP_PROFILE` | unset | Active MCP profile name (server-side only) |
|
||||||
|
|
||||||
|
See [webui-deployment.md](webui-deployment.md) for internal-only serving,
|
||||||
|
Cloudflare Access/WARP/VPN guidance, and unsafe bind overrides (#435).
|
||||||
|
|
||||||
## Routes (MVP)
|
## Routes (MVP)
|
||||||
|
|
||||||
@@ -134,10 +141,68 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
|
|||||||
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
||||||
no tokens or MCP restart actions are exposed.
|
no tokens or MCP restart actions are exposed.
|
||||||
|
|
||||||
|
## Deployment boundary (#435)
|
||||||
|
|
||||||
|
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
|
||||||
|
unless `WEBUI_ALLOW_PUBLIC_BIND=1`. Non-loopback hosts log a warning unless
|
||||||
|
`WEBUI_ALLOW_REMOTE_BIND=1`. `GET /health` exposes `deployment` metadata.
|
||||||
|
|
||||||
## Tests
|
## Tests
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_gated_actions.py -q
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_gated_actions.py -q
|
||||||
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py tests/test_webui_worktree_hygiene.py -q
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py tests/test_webui_worktree_hygiene.py -q
|
||||||
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
|
||||||
```
|
|
||||||
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_deployment_boundary.py -q
|
||||||
|
|
||||||
|
## Tests (#436)
|
||||||
|
|
||||||
|
Run the full hermetic web UI suite (all `test_webui_*.py` modules):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
./scripts/test-webui
|
||||||
|
```
|
||||||
|
|
||||||
|
CI / Jenkins multibranch can call the path-filtered gate (runs only when the
|
||||||
|
diff touches `webui/`, `tests/test_webui_*`, or web UI docs/scripts):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
./scripts/ci-webui-check
|
||||||
|
WEBUI_CI_FORCE=1 ./scripts/ci-webui-check # always run
|
||||||
|
```
|
||||||
|
|
||||||
|
`scripts/test-webui` sets `WEBUI_TEST_OFFLINE=1` by default. In that mode the
|
||||||
|
queue, lease, and runtime routes use empty offline snapshots instead of Gitea
|
||||||
|
credentials, so CI can run without MCP daemon credential access. Set
|
||||||
|
`WEBUI_TEST_OFFLINE=0` only when deliberately validating live fetch behavior.
|
||||||
|
|
||||||
|
Or invoke unittest directly:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
python3 -m unittest discover -s tests -p 'test_webui_*.py' -q
|
||||||
|
```
|
||||||
|
|
||||||
|
## Lease visibility (#433)
|
||||||
|
|
||||||
|
`/leases` surfaces read-only lease and collision state: local issue lock file,
|
||||||
|
in-progress claim inventory (#268), reviewer PR lease comments when present
|
||||||
|
(`<!-- mcp-review-lease:v1 -->`, #407), duplicate open PRs per issue (#400),
|
||||||
|
and duplicate local branches per issue. Links to collision-history backend
|
||||||
|
issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
|
||||||
|
|
||||||
|
## Runtime health (#430)
|
||||||
|
|
||||||
|
`/runtime` surfaces read-only MCP/runtime diagnostics for the default registry
|
||||||
|
project: active profile and role kind, authenticated identity (when credentials
|
||||||
|
are available), config model/mode, local vs remote `master` SHA sync, shell
|
||||||
|
health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
|
||||||
|
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
||||||
|
no tokens or MCP restart actions are exposed.
|
||||||
|
|
||||||
|
## Tests
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py tests/test_webui_worktree_hygiene.py -q
|
||||||
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
|
||||||
|
```
|
||||||
|
|||||||
Executable
+53
@@ -0,0 +1,53 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Path-filtered CI gate for the internal web UI (#436).
|
||||||
|
# Runs the hermetic web UI unittest suite when a change touches UI surfaces.
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||||
|
repo_root="$(cd "$script_dir/.." && pwd)"
|
||||||
|
cd "$repo_root"
|
||||||
|
|
||||||
|
if [[ "${WEBUI_CI_FORCE:-}" == "1" ]]; then
|
||||||
|
exec "$script_dir/test-webui"
|
||||||
|
fi
|
||||||
|
|
||||||
|
base_ref="${WEBUI_CI_BASE_REF:-}"
|
||||||
|
if [[ -z "$base_ref" ]]; then
|
||||||
|
if git rev-parse --verify prgs/master >/dev/null 2>&1; then
|
||||||
|
base_ref="$(git merge-base HEAD prgs/master 2>/dev/null || true)"
|
||||||
|
fi
|
||||||
|
if [[ -z "$base_ref" ]]; then
|
||||||
|
base_ref="${GITHUB_BASE_REF:-${CHANGE_TARGET:-master}}"
|
||||||
|
if git rev-parse --verify "origin/$base_ref" >/dev/null 2>&1; then
|
||||||
|
base_ref="$(git merge-base HEAD "origin/$base_ref")"
|
||||||
|
elif git rev-parse --verify "$base_ref" >/dev/null 2>&1; then
|
||||||
|
base_ref="$(git merge-base HEAD "$base_ref")"
|
||||||
|
else
|
||||||
|
base_ref="HEAD~1"
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
|
||||||
|
changed="$(git diff --name-only "$base_ref" HEAD 2>/dev/null || true)"
|
||||||
|
if [[ -z "$changed" ]]; then
|
||||||
|
echo "ci-webui-check: no changed files vs $base_ref; skipping web UI suite"
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
|
||||||
|
python_bin="${WEBUI_TEST_PYTHON:-python3}"
|
||||||
|
if [[ -x "$repo_root/venv/bin/python" ]]; then
|
||||||
|
python_bin="$repo_root/venv/bin/python"
|
||||||
|
fi
|
||||||
|
|
||||||
|
if printf '%s\n' "$changed" | "$python_bin" -c "
|
||||||
|
import sys
|
||||||
|
from webui.ci_paths import should_run_webui_ci
|
||||||
|
paths = [line.strip() for line in sys.stdin if line.strip()]
|
||||||
|
raise SystemExit(0 if should_run_webui_ci(paths) else 1)
|
||||||
|
"; then
|
||||||
|
echo "ci-webui-check: web UI surface changed; running suite"
|
||||||
|
exec "$script_dir/test-webui"
|
||||||
|
fi
|
||||||
|
|
||||||
|
echo "ci-webui-check: no web UI paths in diff; skipping suite"
|
||||||
|
exit 0
|
||||||
Executable
+14
@@ -0,0 +1,14 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
set -euo pipefail
|
||||||
|
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||||
|
repo_root="$(cd "$script_dir/.." && pwd)"
|
||||||
|
cd "$repo_root"
|
||||||
|
|
||||||
|
python_bin="${WEBUI_TEST_PYTHON:-python3}"
|
||||||
|
if [[ -x "$repo_root/venv/bin/python" ]]; then
|
||||||
|
python_bin="$repo_root/venv/bin/python"
|
||||||
|
fi
|
||||||
|
|
||||||
|
pattern="${WEBUI_TEST_PATTERN:-test_webui_*.py}"
|
||||||
|
export WEBUI_TEST_OFFLINE="${WEBUI_TEST_OFFLINE:-1}"
|
||||||
|
exec "$python_bin" -m unittest discover -s tests -p "$pattern" "$@"
|
||||||
@@ -0,0 +1,90 @@
|
|||||||
|
"""Tests for web UI CI path-filter gate (#436)."""
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
from webui.ci_paths import should_run_webui_ci, touches_webui_surface
|
||||||
|
from webui.lease_loader import load_lease_snapshot
|
||||||
|
from webui.queue_loader import load_queue_snapshot
|
||||||
|
from webui.runtime_health import load_runtime_snapshot
|
||||||
|
|
||||||
|
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||||
|
|
||||||
|
|
||||||
|
class TestCiPathMatching(unittest.TestCase):
|
||||||
|
def test_positive_paths(self):
|
||||||
|
for path in (
|
||||||
|
"webui/app.py",
|
||||||
|
"tests/test_webui_skeleton.py",
|
||||||
|
"docs/webui-local-dev.md",
|
||||||
|
"scripts/test-webui",
|
||||||
|
):
|
||||||
|
with self.subTest(path=path):
|
||||||
|
self.assertTrue(touches_webui_surface(path))
|
||||||
|
|
||||||
|
def test_negative_paths(self):
|
||||||
|
for path in ("mcp_server.py", "tests/test_mcp_server.py", "README.md"):
|
||||||
|
with self.subTest(path=path):
|
||||||
|
self.assertFalse(touches_webui_surface(path))
|
||||||
|
|
||||||
|
def test_should_run_aggregate(self):
|
||||||
|
self.assertTrue(should_run_webui_ci(["README.md", "webui/layout.py"]))
|
||||||
|
self.assertFalse(should_run_webui_ci(["mcp_server.py", "gitea_auth.py"]))
|
||||||
|
|
||||||
|
|
||||||
|
class TestCiRunnerScript(unittest.TestCase):
|
||||||
|
def test_test_webui_smoke(self):
|
||||||
|
script = _REPO_ROOT / "scripts" / "test-webui"
|
||||||
|
result = subprocess.run(
|
||||||
|
["bash", str(script), "-q"],
|
||||||
|
cwd=_REPO_ROOT,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=60,
|
||||||
|
env={
|
||||||
|
**os.environ,
|
||||||
|
"WEBUI_TEST_PATTERN": "test_webui_skeleton.py",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
self.assertEqual(result.returncode, 0, result.stderr or result.stdout)
|
||||||
|
|
||||||
|
|
||||||
|
class TestOfflineWebuiCiMode(unittest.TestCase):
|
||||||
|
def test_offline_mode_avoids_live_queue_auth(self):
|
||||||
|
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
|
||||||
|
with mock.patch(
|
||||||
|
"webui.queue_loader.get_auth_header",
|
||||||
|
side_effect=AssertionError("live auth should not run"),
|
||||||
|
):
|
||||||
|
snapshot = load_queue_snapshot()
|
||||||
|
self.assertIsNone(snapshot.fetch_error)
|
||||||
|
self.assertEqual(snapshot.prs, ())
|
||||||
|
self.assertEqual(snapshot.issues, ())
|
||||||
|
|
||||||
|
def test_offline_mode_avoids_live_lease_auth(self):
|
||||||
|
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
|
||||||
|
with mock.patch(
|
||||||
|
"webui.lease_loader.get_auth_header",
|
||||||
|
side_effect=AssertionError("live auth should not run"),
|
||||||
|
):
|
||||||
|
snapshot = load_lease_snapshot()
|
||||||
|
self.assertIsNone(snapshot.fetch_error)
|
||||||
|
self.assertEqual(snapshot.reviewer_leases, ())
|
||||||
|
|
||||||
|
def test_offline_mode_avoids_live_runtime_identity(self):
|
||||||
|
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
|
||||||
|
with mock.patch(
|
||||||
|
"webui.runtime_health.get_auth_header",
|
||||||
|
side_effect=AssertionError("live auth should not run"),
|
||||||
|
):
|
||||||
|
snapshot = load_runtime_snapshot()
|
||||||
|
self.assertEqual(snapshot.identity_error, "offline web UI test mode")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -0,0 +1,106 @@
|
|||||||
|
"""Tests for web UI deployment and auth boundary (#435)."""
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
from starlette.testclient import TestClient
|
||||||
|
|
||||||
|
from webui.app import create_app
|
||||||
|
from webui.deployment_boundary import (
|
||||||
|
assess_bind_host,
|
||||||
|
runtime_assumptions,
|
||||||
|
scan_text_for_client_secrets,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestBindAssessment(unittest.TestCase):
|
||||||
|
def test_loopback_is_safe(self):
|
||||||
|
for host in ("127.0.0.1", "localhost", "::1"):
|
||||||
|
with self.subTest(host=host):
|
||||||
|
result = assess_bind_host(host)
|
||||||
|
self.assertEqual(result.disposition, "safe")
|
||||||
|
|
||||||
|
def test_all_interfaces_refused_by_default(self):
|
||||||
|
for host in ("0.0.0.0", "::", "*"):
|
||||||
|
with self.subTest(host=host):
|
||||||
|
result = assess_bind_host(host)
|
||||||
|
self.assertEqual(result.disposition, "refuse")
|
||||||
|
|
||||||
|
def test_all_interfaces_allowed_with_override(self):
|
||||||
|
prior = os.environ.pop("WEBUI_ALLOW_PUBLIC_BIND", None)
|
||||||
|
try:
|
||||||
|
os.environ["WEBUI_ALLOW_PUBLIC_BIND"] = "1"
|
||||||
|
result = assess_bind_host("0.0.0.0")
|
||||||
|
self.assertEqual(result.disposition, "warn")
|
||||||
|
finally:
|
||||||
|
os.environ.pop("WEBUI_ALLOW_PUBLIC_BIND", None)
|
||||||
|
if prior is not None:
|
||||||
|
os.environ["WEBUI_ALLOW_PUBLIC_BIND"] = prior
|
||||||
|
|
||||||
|
def test_non_loopback_warns_without_override(self):
|
||||||
|
prior = os.environ.pop("WEBUI_ALLOW_REMOTE_BIND", None)
|
||||||
|
try:
|
||||||
|
os.environ.pop("WEBUI_ALLOW_REMOTE_BIND", None)
|
||||||
|
result = assess_bind_host("192.168.1.10")
|
||||||
|
self.assertEqual(result.disposition, "warn")
|
||||||
|
finally:
|
||||||
|
if prior is not None:
|
||||||
|
os.environ["WEBUI_ALLOW_REMOTE_BIND"] = prior
|
||||||
|
|
||||||
|
def test_runtime_assumptions_exclude_secrets(self):
|
||||||
|
assumptions = runtime_assumptions()
|
||||||
|
blob = " ".join(str(v) for v in assumptions.values())
|
||||||
|
self.assertNotIn("GITEA_TOKEN", blob)
|
||||||
|
self.assertIn("internal-operator-console", assumptions["deployment_mode"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestDeploymentRoutes(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.client = TestClient(create_app(bind_host="127.0.0.1"))
|
||||||
|
|
||||||
|
def test_health_includes_deployment_metadata(self):
|
||||||
|
response = self.client.get("/health")
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
data = response.json()
|
||||||
|
self.assertIn("deployment", data)
|
||||||
|
deployment = data["deployment"]
|
||||||
|
self.assertEqual(deployment["mode"], "internal-operator-console")
|
||||||
|
self.assertEqual(deployment["mvp_auth"], "none")
|
||||||
|
self.assertEqual(deployment["bind"]["disposition"], "safe")
|
||||||
|
self.assertIn("runtime_assumptions", deployment)
|
||||||
|
|
||||||
|
def test_rendered_pages_contain_no_embedded_secrets(self):
|
||||||
|
for path in ("/", "/projects", "/prompts", "/queue"):
|
||||||
|
with self.subTest(path=path):
|
||||||
|
text = self.client.get(path).text
|
||||||
|
self.assertEqual(scan_text_for_client_secrets(text), [])
|
||||||
|
|
||||||
|
|
||||||
|
class TestStartupRefusal(unittest.TestCase):
|
||||||
|
def test_refuses_all_interface_bind(self):
|
||||||
|
env = {**os.environ, "WEBUI_HOST": "0.0.0.0"}
|
||||||
|
env.pop("WEBUI_ALLOW_PUBLIC_BIND", None)
|
||||||
|
repo_root = Path(__file__).resolve().parent.parent
|
||||||
|
result = subprocess.run(
|
||||||
|
[sys.executable, "-m", "webui"],
|
||||||
|
cwd=repo_root,
|
||||||
|
env=env,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=3,
|
||||||
|
)
|
||||||
|
self.assertEqual(result.returncode, 1)
|
||||||
|
|
||||||
|
|
||||||
|
class TestClientSecretScanner(unittest.TestCase):
|
||||||
|
def test_detects_token_like_content(self):
|
||||||
|
findings = scan_text_for_client_secrets("var x = GITEA_TOKEN_PRGS")
|
||||||
|
self.assertTrue(findings)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -0,0 +1,156 @@
|
|||||||
|
"""Consolidated MVP coverage for the internal web UI (#436)."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
from starlette.testclient import TestClient
|
||||||
|
from starlette.routing import Route
|
||||||
|
|
||||||
|
from webui.app import create_app
|
||||||
|
|
||||||
|
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||||
|
|
||||||
|
# HTML pages and JSON exports registered on master MVP.
|
||||||
|
MVP_GET_ROUTES: tuple[tuple[str, str | tuple[str, ...]], ...] = (
|
||||||
|
("/", "Operator console"),
|
||||||
|
("/health", "mcp-control-plane-webui"),
|
||||||
|
("/queue", "Live queue"),
|
||||||
|
("/api/queue", ("prs", "issues")),
|
||||||
|
("/projects", "Gitea-Tools"),
|
||||||
|
("/api/projects", "projects"),
|
||||||
|
("/prompts", "Prompt library"),
|
||||||
|
("/api/prompts", "prompts"),
|
||||||
|
("/runtime", "Runtime"),
|
||||||
|
("/audit", "Audit"),
|
||||||
|
("/worktrees", "Worktrees"),
|
||||||
|
("/leases", "Leases"),
|
||||||
|
)
|
||||||
|
|
||||||
|
MUTATION_METHODS = ("POST", "PUT", "PATCH", "DELETE")
|
||||||
|
|
||||||
|
|
||||||
|
def _import_optional(module_name: str):
|
||||||
|
try:
|
||||||
|
return importlib.import_module(module_name)
|
||||||
|
except ImportError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
class TestWebuiServerStartup(unittest.TestCase):
|
||||||
|
def test_create_app_imports_cleanly(self):
|
||||||
|
app = create_app()
|
||||||
|
self.assertIsNotNone(app)
|
||||||
|
|
||||||
|
def test_main_module_imports(self):
|
||||||
|
import webui.__main__ as main_mod
|
||||||
|
|
||||||
|
self.assertTrue(callable(main_mod.main))
|
||||||
|
|
||||||
|
|
||||||
|
class TestWebuiRouteRendering(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.client = TestClient(create_app())
|
||||||
|
|
||||||
|
def test_all_mvp_get_routes_render(self):
|
||||||
|
for path, needle in MVP_GET_ROUTES:
|
||||||
|
with self.subTest(path=path):
|
||||||
|
response = self.client.get(path)
|
||||||
|
self.assertEqual(response.status_code, 200, path)
|
||||||
|
if path == "/health":
|
||||||
|
assert isinstance(needle, str)
|
||||||
|
self.assertEqual(response.json()["service"], needle)
|
||||||
|
elif path.startswith("/api/"):
|
||||||
|
data = response.json()
|
||||||
|
if isinstance(needle, tuple):
|
||||||
|
for key in needle:
|
||||||
|
self.assertIn(key, data)
|
||||||
|
else:
|
||||||
|
self.assertIn(needle, data)
|
||||||
|
else:
|
||||||
|
assert isinstance(needle, str)
|
||||||
|
self.assertIn(needle, response.text)
|
||||||
|
|
||||||
|
def test_registered_routes_include_mvp_paths(self):
|
||||||
|
app = create_app()
|
||||||
|
get_paths = {
|
||||||
|
route.path
|
||||||
|
for route in app.routes
|
||||||
|
if isinstance(route, Route) and "GET" in route.methods
|
||||||
|
}
|
||||||
|
for path, _ in MVP_GET_ROUTES:
|
||||||
|
with self.subTest(path=path):
|
||||||
|
self.assertIn(path, get_paths)
|
||||||
|
|
||||||
|
|
||||||
|
class TestWebuiReadOnlyGuards(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.client = TestClient(create_app())
|
||||||
|
|
||||||
|
def test_mutation_methods_rejected_on_core_paths(self):
|
||||||
|
paths = ("/health", "/queue", "/projects", "/prompts", "/api/queue")
|
||||||
|
for path in paths:
|
||||||
|
for method in MUTATION_METHODS:
|
||||||
|
with self.subTest(path=path, method=method):
|
||||||
|
response = self.client.request(method, path)
|
||||||
|
self.assertEqual(response.status_code, 405)
|
||||||
|
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||||
|
|
||||||
|
|
||||||
|
class TestWebuiOptionalModules(unittest.TestCase):
|
||||||
|
"""Exercise child-issue modules when present on the branch under test."""
|
||||||
|
|
||||||
|
def test_audit_validator_basics_when_present(self):
|
||||||
|
mod = _import_optional("webui.audit_validator")
|
||||||
|
if mod is None:
|
||||||
|
self.skipTest("webui.audit_validator not merged on this branch")
|
||||||
|
report = "## Controller Handoff\n- Task: review PR #1\n"
|
||||||
|
self.assertEqual(mod.infer_task_kind(report), "review_pr")
|
||||||
|
result = mod.audit_report(report)
|
||||||
|
self.assertIn("grade", result)
|
||||||
|
|
||||||
|
def test_gated_actions_fail_closed_when_present(self):
|
||||||
|
mod = _import_optional("webui.gated_actions")
|
||||||
|
if mod is None:
|
||||||
|
self.skipTest("webui.gated_actions not merged on this branch")
|
||||||
|
registry = mod.build_action_registry()
|
||||||
|
for action in registry.actions:
|
||||||
|
with self.subTest(action=action.action_id):
|
||||||
|
self.assertFalse(action.enabled)
|
||||||
|
result = mod.attempt_action("merge_pr", pr_number=1)
|
||||||
|
self.assertFalse(result["success"])
|
||||||
|
|
||||||
|
def test_deployment_boundary_when_present(self):
|
||||||
|
mod = _import_optional("webui.deployment_boundary")
|
||||||
|
if mod is None:
|
||||||
|
self.skipTest("webui.deployment_boundary not merged on this branch")
|
||||||
|
assessment = mod.assess_bind_host("0.0.0.0")
|
||||||
|
self.assertEqual(assessment.disposition, "refuse")
|
||||||
|
|
||||||
|
|
||||||
|
class TestWebuiTestInventory(unittest.TestCase):
|
||||||
|
def test_webui_test_modules_exist(self):
|
||||||
|
modules = sorted(_REPO_ROOT.glob("tests/test_webui_*.py"))
|
||||||
|
names = [path.name for path in modules]
|
||||||
|
self.assertIn("test_webui_skeleton.py", names)
|
||||||
|
self.assertIn("test_webui_project_registry.py", names)
|
||||||
|
self.assertIn("test_webui_prompt_library.py", names)
|
||||||
|
self.assertIn("test_webui_queue_dashboard.py", names)
|
||||||
|
self.assertGreaterEqual(len(names), 5)
|
||||||
|
|
||||||
|
|
||||||
|
class TestWebuiCiScripts(unittest.TestCase):
|
||||||
|
def test_runner_scripts_exist(self):
|
||||||
|
for name in ("test-webui", "ci-webui-check"):
|
||||||
|
script = _REPO_ROOT / "scripts" / name
|
||||||
|
self.assertTrue(script.is_file(), name)
|
||||||
|
self.assertTrue(os.access(script, os.X_OK), name)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -206,8 +206,9 @@ class TestQueueRoutes(unittest.TestCase):
|
|||||||
self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
|
self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
|
||||||
|
|
||||||
def test_queue_fail_closed_without_credentials(self):
|
def test_queue_fail_closed_without_credentials(self):
|
||||||
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
|
with mock.patch.dict("os.environ", {"WEBUI_TEST_OFFLINE": "0"}):
|
||||||
snapshot = load_queue_snapshot()
|
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
|
||||||
|
snapshot = load_queue_snapshot()
|
||||||
self.assertIsNotNone(snapshot.fetch_error)
|
self.assertIsNotNone(snapshot.fetch_error)
|
||||||
self.assertEqual(len(snapshot.prs), 0)
|
self.assertEqual(len(snapshot.prs), 0)
|
||||||
|
|
||||||
@@ -285,4 +286,4 @@ class TestQueueFailClosedUx(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -22,6 +22,8 @@ class TestWebuiSkeleton(unittest.TestCase):
|
|||||||
self.assertEqual(data["service"], "mcp-control-plane-webui")
|
self.assertEqual(data["service"], "mcp-control-plane-webui")
|
||||||
self.assertEqual(data["mode"], "read-only-mvp")
|
self.assertEqual(data["mode"], "read-only-mvp")
|
||||||
self.assertIn("timestamp", data)
|
self.assertIn("timestamp", data)
|
||||||
|
self.assertIn("deployment", data)
|
||||||
|
self.assertEqual(data["deployment"]["mode"], "internal-operator-console")
|
||||||
|
|
||||||
def test_home_renders(self):
|
def test_home_renders(self):
|
||||||
response = self.client.get("/")
|
response = self.client.get("/")
|
||||||
|
|||||||
+19
-2
@@ -2,18 +2,35 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
|
|
||||||
import uvicorn
|
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:
|
def main() -> None:
|
||||||
host = os.environ.get("WEBUI_HOST", "127.0.0.1")
|
host = os.environ.get("WEBUI_HOST", "127.0.0.1")
|
||||||
port = int(os.environ.get("WEBUI_PORT", "8765"))
|
port = int(os.environ.get("WEBUI_PORT", "8765"))
|
||||||
|
# Validate bind host before importing the full app stack so refuse paths
|
||||||
|
# fail closed quickly (tests and operators must not hang on create_app).
|
||||||
|
_validate_bind_host(host)
|
||||||
|
from webui.app import create_app
|
||||||
|
|
||||||
uvicorn.run(create_app(), host=host, port=port, log_level="info")
|
uvicorn.run(create_app(), host=host, port=port, log_level="info")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
|
|||||||
+11
-3
@@ -9,6 +9,7 @@ from starlette.requests import Request
|
|||||||
from starlette.responses import HTMLResponse, JSONResponse, Response
|
from starlette.responses import HTMLResponse, JSONResponse, Response
|
||||||
from starlette.routing import Route
|
from starlette.routing import Route
|
||||||
|
|
||||||
|
from webui.deployment_boundary import deployment_snapshot
|
||||||
from webui.layout import render_page
|
from webui.layout import render_page
|
||||||
from webui.project_registry import find_project, load_registry, registry_to_dict
|
from webui.project_registry import find_project, load_registry, registry_to_dict
|
||||||
from webui.project_views import render_project_detail, render_projects_list
|
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:
|
async def health(_request: Request) -> JSONResponse:
|
||||||
|
bind_host = _request.app.state.webui_bind_host
|
||||||
return JSONResponse({
|
return JSONResponse({
|
||||||
"status": "ok",
|
"status": "ok",
|
||||||
"service": "mcp-control-plane-webui",
|
"service": "mcp-control-plane-webui",
|
||||||
"mode": "read-only-mvp",
|
"mode": "read-only-mvp",
|
||||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
"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)
|
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."""
|
"""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,
|
debug=False,
|
||||||
routes=[
|
routes=[
|
||||||
Route("/", home, methods=["GET"]),
|
Route("/", home, methods=["GET"]),
|
||||||
@@ -287,4 +293,6 @@ def create_app() -> Starlette:
|
|||||||
Route("/api/leases", api_leases, methods=["GET"]),
|
Route("/api/leases", api_leases, methods=["GET"]),
|
||||||
],
|
],
|
||||||
exception_handlers={405: method_not_allowed},
|
exception_handlers={405: method_not_allowed},
|
||||||
)
|
)
|
||||||
|
app.state.webui_bind_host = resolved_bind
|
||||||
|
return app
|
||||||
@@ -0,0 +1,20 @@
|
|||||||
|
"""Path filters for web UI CI gating (#436)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
from collections.abc import Iterable
|
||||||
|
|
||||||
|
_WEBUI_TOUCH_RE = re.compile(
|
||||||
|
r"^(?:webui/|tests/test_webui_|docs/webui|scripts/(?:run-webui|test-webui|ci-webui-check)$)"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def touches_webui_surface(path: str) -> bool:
|
||||||
|
"""Return True when *path* is a web UI surface file."""
|
||||||
|
return bool(_WEBUI_TOUCH_RE.match(path.strip()))
|
||||||
|
|
||||||
|
|
||||||
|
def should_run_webui_ci(changed_paths: Iterable[str]) -> bool:
|
||||||
|
"""Return True when any changed path should trigger the web UI test suite."""
|
||||||
|
return any(touches_webui_surface(path) for path in changed_paths)
|
||||||
@@ -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."
|
||||||
|
),
|
||||||
|
}
|
||||||
+24
-5
@@ -228,10 +228,29 @@ def load_lease_snapshot(
|
|||||||
)
|
)
|
||||||
|
|
||||||
host = _host_from_url(project.remote_host)
|
host = _host_from_url(project.remote_host)
|
||||||
pr_fetch = fetch_prs or _fetch_prs
|
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
|
||||||
issue_fetch = fetch_issues or _fetch_issues
|
"1",
|
||||||
comment_fetch = fetch_comments or _fetch_comments
|
"true",
|
||||||
using_live = fetch_prs is None or fetch_issues is None
|
"yes",
|
||||||
|
}
|
||||||
|
|
||||||
|
def _empty_pr_fetch(*_args, **_kwargs):
|
||||||
|
return [], None
|
||||||
|
|
||||||
|
def _empty_issue_fetch(*_args, **_kwargs):
|
||||||
|
return [], None
|
||||||
|
|
||||||
|
def _empty_comment_fetch(*_args, **_kwargs):
|
||||||
|
return []
|
||||||
|
|
||||||
|
pr_fetch = fetch_prs or (_empty_pr_fetch if offline_test else _fetch_prs)
|
||||||
|
issue_fetch = fetch_issues or (
|
||||||
|
_empty_issue_fetch if offline_test else _fetch_issues
|
||||||
|
)
|
||||||
|
comment_fetch = fetch_comments or (
|
||||||
|
_empty_comment_fetch if offline_test else _fetch_comments
|
||||||
|
)
|
||||||
|
using_live = not offline_test and (fetch_prs is None or fetch_issues is None)
|
||||||
auth = get_auth_header(host) if using_live else "test-auth"
|
auth = get_auth_header(host) if using_live else "test-auth"
|
||||||
|
|
||||||
if using_live and not auth:
|
if using_live and not auth:
|
||||||
@@ -328,4 +347,4 @@ def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
|
|||||||
],
|
],
|
||||||
"collision_history": list(snapshot.collision_history),
|
"collision_history": list(snapshot.collision_history),
|
||||||
"fetch_error": snapshot.fetch_error,
|
"fetch_error": snapshot.fetch_error,
|
||||||
}
|
}
|
||||||
|
|||||||
+19
-4
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
import re
|
import re
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
@@ -268,9 +269,23 @@ def load_queue_snapshot(
|
|||||||
)
|
)
|
||||||
|
|
||||||
host = _host_from_url(project.remote_host)
|
host = _host_from_url(project.remote_host)
|
||||||
pr_fetch = fetch_prs or _fetch_prs
|
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
|
||||||
issue_fetch = fetch_issues or _fetch_issues
|
"1",
|
||||||
using_live_fetch = fetch_prs is None or fetch_issues is None
|
"true",
|
||||||
|
"yes",
|
||||||
|
}
|
||||||
|
|
||||||
|
def _empty_fetch(*_args, **_kwargs):
|
||||||
|
return [], _pagination_from_pages(
|
||||||
|
per_page=50,
|
||||||
|
pages_fetched=1,
|
||||||
|
returned_count=0,
|
||||||
|
is_final_page=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
|
||||||
|
issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
|
||||||
|
using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
|
||||||
auth = get_auth_header(host) if using_live_fetch else "test-auth"
|
auth = get_auth_header(host) if using_live_fetch else "test-auth"
|
||||||
if using_live_fetch and not auth:
|
if using_live_fetch and not auth:
|
||||||
return QueueSnapshot(
|
return QueueSnapshot(
|
||||||
@@ -382,4 +397,4 @@ def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
|
|||||||
"prs": _page(snapshot.pr_pagination),
|
"prs": _page(snapshot.pr_pagination),
|
||||||
"issues": _page(snapshot.issue_pagination),
|
"issues": _page(snapshot.issue_pagination),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|||||||
+18
-3
@@ -217,6 +217,11 @@ def load_runtime_snapshot(
|
|||||||
repo = _repo_root()
|
repo = _repo_root()
|
||||||
host = _host_from_url(project.remote_host)
|
host = _host_from_url(project.remote_host)
|
||||||
remote = _remote_name_for_host(host)
|
remote = _remote_name_for_host(host)
|
||||||
|
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
|
||||||
|
"1",
|
||||||
|
"true",
|
||||||
|
"yes",
|
||||||
|
}
|
||||||
profile = get_profile()
|
profile = get_profile()
|
||||||
allowed = profile.get("allowed_operations") or []
|
allowed = profile.get("allowed_operations") or []
|
||||||
forbidden = profile.get("forbidden_operations") or []
|
forbidden = profile.get("forbidden_operations") or []
|
||||||
@@ -247,10 +252,20 @@ def load_runtime_snapshot(
|
|||||||
fetch_error=str(exc),
|
fetch_error=str(exc),
|
||||||
)
|
)
|
||||||
|
|
||||||
identity_fn = resolve_username or _resolve_username
|
identity_fn = resolve_username
|
||||||
|
if identity_fn is None:
|
||||||
|
if offline_test:
|
||||||
|
identity_fn = lambda _host: (None, "offline web UI test mode")
|
||||||
|
else:
|
||||||
|
identity_fn = _resolve_username
|
||||||
username, identity_error = identity_fn(host)
|
username, identity_error = identity_fn(host)
|
||||||
|
|
||||||
git_fn = git_sync or _git_sync_status
|
git_fn = git_sync
|
||||||
|
if git_fn is None:
|
||||||
|
if offline_test:
|
||||||
|
git_fn = lambda _repo, _remote, _branch: (None, None, None)
|
||||||
|
else:
|
||||||
|
git_fn = _git_sync_status
|
||||||
remote_ref = "prgs" if remote == "prgs" else "origin"
|
remote_ref = "prgs" if remote == "prgs" else "origin"
|
||||||
local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch)
|
local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch)
|
||||||
|
|
||||||
@@ -317,4 +332,4 @@ def snapshot_to_dict(snapshot: RuntimeSnapshot) -> dict[str, Any]:
|
|||||||
"schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
|
"schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
|
||||||
"restart_guidance": snapshot.restart_guidance,
|
"restart_guidance": snapshot.restart_guidance,
|
||||||
"fetch_error": snapshot.fetch_error,
|
"fetch_error": snapshot.fetch_error,
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user