gitea-admin/nocobase-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: add secure personal key setup

Browse Source
This commit is contained in:
nac
2026-08-14 15:42:02 +00:00
parent 67d7ff6608
commit 9ad9474226
11 changed files with 434 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/agent_install.py
+223 -1
View File
@@ -21,12 +21,14 @@ from collections.abc import Callable, Sequence
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit, urlunsplit
MIN_PYTHON = (3, 12)
DEFAULT_ENV_NAME = "nb-mcp"
SERVER_NAME = "nocobase-crm"
REQUIRED_PROJECT_FILES = ("pyproject.toml", "uv.lock", "config.example.json")
ENV_NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
ANSI_ESCAPE_PATTERN = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
Which = Callable[[str], str | None]
Runner = Callable[..., subprocess.CompletedProcess[str]]
@@ -200,6 +202,213 @@ def set_config_env(project_root: str | Path | None, env_name: str) -> dict[str,
}
def normalize_api_base_url(value: str) -> str:
raw = value.strip()
if not raw:
raise InstallerError("NocoBase API base URL is required")
parsed = urlsplit(raw)
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
raise InstallerError("NocoBase URL must be an absolute http:// or https:// URL")
if parsed.username or parsed.password or parsed.query or parsed.fragment:
raise InstallerError("NocoBase URL must not contain credentials, a query string, or a fragment")
hostname = parsed.hostname.lower()
is_local = hostname == "localhost" or hostname == "::1" or hostname.startswith("127.")
if parsed.scheme != "https" and not is_local:
raise InstallerError("A remote NocoBase API Key may only be sent to an https:// endpoint")
path = parsed.path.rstrip("/")
if not path.endswith("/api"):
path = f"{path}/api" if path else "/api"
return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
def _parse_cli_json(output: str) -> dict[str, Any]:
decoder = json.JSONDecoder()
candidates: list[tuple[int, int, dict[str, Any]]] = []
for start, character in enumerate(output):
if character != "{":
continue
try:
value, end = decoder.raw_decode(output, start)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
candidates.append((start, end, value))
outer_candidates = [
candidate
for candidate in candidates
if not any(
other_start <= candidate[0]
and other_end >= candidate[1]
and (other_start, other_end) != (candidate[0], candidate[1])
for other_start, other_end, _value in candidates
)
]
if len(outer_candidates) != 1:
raise InstallerError("nb env info did not return one JSON object")
return outer_candidates[0][2]
def _nb_command_environment(nb_bin: str) -> dict[str, str]:
environment = os.environ.copy()
environment["NB_CLI_LOG_DISABLED"] = "1"
environment["NB_SKIP_STARTUP_UPDATE"] = "1"
nb_path = Path(nb_bin)
if nb_path.is_absolute():
nb_dir = str(nb_path.parent)
existing_path = environment.get("PATH", "")
parts = [part for part in existing_path.split(os.pathsep) if part]
if nb_dir not in parts:
environment["PATH"] = nb_dir + (os.pathsep + existing_path if existing_path else "")
return environment
def _read_nb_environment(
root: Path,
env_name: str,
*,
nb_bin: str,
runner: Runner,
) -> dict[str, Any] | None:
command = [nb_bin, "env", "info", env_name, "--json"]
try:
result = runner(
command,
cwd=str(root),
env=_nb_command_environment(nb_bin),
capture_output=True,
text=True,
check=False,
)
except OSError as error:
raise InstallerError(f"Could not execute nb env info: {error}") from error
if result.returncode == 0:
return _parse_cli_json(result.stdout)
diagnostic = ANSI_ESCAPE_PATTERN.sub("", f"{result.stdout}\n{result.stderr}")
if "is not configured" in diagnostic:
return None
raise InstallerError(
f"nb env info failed with exit code {result.returncode}; inspect the nb CLI diagnostic locally"
)
def _api_details(payload: dict[str, Any]) -> tuple[str | None, str | None]:
api = payload.get("api")
auth = api.get("auth") if isinstance(api, dict) else None
api_base_url = api.get("apiBaseUrl") if isinstance(api, dict) else None
auth_type = auth.get("type") if isinstance(auth, dict) else None
normalized_url = api_base_url.strip().rstrip("/") if isinstance(api_base_url, str) else None
normalized_type = auth_type.strip() if isinstance(auth_type, str) else None
return normalized_url, normalized_type
def _suggest_env_name(api_base_url: str) -> str:
hostname = urlsplit(api_base_url).hostname or "server"
suffix = re.sub(r"[^a-z0-9]+", "-", hostname.lower()).strip("-")[:32] or "server"
return f"{DEFAULT_ENV_NAME}-{suffix}"
def connect_with_personal_key(
project_root: str | Path | None,
api_base_url: str,
*,
env_name: str = DEFAULT_ENV_NAME,
runner: Runner = subprocess.run,
which: Which = shutil.which,
stdin_isatty: Callable[[], bool] = sys.stdin.isatty,
) -> dict[str, Any]:
"""Bind a personal API Key through nb's private TTY prompt.
The key is never accepted by this function, placed in argv, inherited from
the process environment, captured, printed, or returned. The child nb CLI
reads it directly from the user-controlled terminal.
"""
if not ENV_NAME_PATTERN.fullmatch(env_name):
raise InstallerError(
"Environment name must start with a letter or digit and contain only letters, digits, dot, underscore, or dash"
)
if not stdin_isatty():
raise InstallerError(
"Secure API Key entry requires a user-controlled local terminal. "
"Do not paste the Key into Agent chat or pass it with --access-token."
)
root = _project_root(project_root)
normalized_url = normalize_api_base_url(api_base_url)
discovered_nb = which("nb")
if not discovered_nb:
raise InstallerError("nb was not found on PATH")
nb_bin = os.path.abspath(discovered_nb)
existing = _read_nb_environment(root, env_name, nb_bin=nb_bin, runner=runner)
if existing is None:
interactive_command = [
nb_bin,
"env",
"add",
env_name,
"--api-base-url",
normalized_url,
"--auth-type",
"token",
]
action = "created"
else:
existing_url, _auth_type = _api_details(existing)
if existing_url != normalized_url.rstrip("/"):
suggestion = _suggest_env_name(normalized_url)
raise InstallerError(
f"Environment {env_name!r} already targets {existing_url or 'an unconfigured endpoint'!r}; "
f"use a new name such as {suggestion!r} instead of overwriting it"
)
interactive_command = [nb_bin, "env", "auth", env_name, "--auth-type", "token"]
action = "reauthenticated"
try:
result = runner(
interactive_command,
cwd=str(root),
env=_nb_command_environment(nb_bin),
text=True,
check=False,
)
except OSError as error:
raise InstallerError(f"Could not execute the interactive nb Key prompt: {error}") from error
if result.returncode != 0:
raise InstallerError(
f"Interactive API Key authentication failed with exit code {result.returncode}; "
"revoke the exposed or invalid Key if necessary and retry through the installer"
)
verified = _read_nb_environment(root, env_name, nb_bin=nb_bin, runner=runner)
if verified is None:
raise InstallerError("nb reported success but the saved environment could not be read back")
verified_url, auth_type = _api_details(verified)
if verified_url != normalized_url.rstrip("/") or auth_type != "token":
raise InstallerError("The saved nb environment did not verify as token authentication for the requested URL")
list_result = runner(
[nb_bin, "env", "list"],
cwd=str(root),
env=_nb_command_environment(nb_bin),
capture_output=True,
text=True,
check=False,
)
if list_result.returncode != 0:
raise InstallerError("API Key was saved, but nb env list readback failed")
config_result = set_config_env(root, env_name)
return {
"ok": True,
"action": action,
"env": env_name,
"apiBaseUrl": normalized_url,
"authType": "token",
"keyExposure": "private-tty-only",
"configChanged": config_result["changed"],
"configBackupPath": config_result["backupPath"],
}
def render_client_config(
project_root: str | Path | None = None,
*,
@@ -298,7 +507,7 @@ def doctor_project(
"projectRoot": str(root),
"checks": [asdict(check) for check in checks],
"remainingVerification": [
"Run nb env info <configured-env> in its default redacted mode.",
"Have the installation Agent verify the saved env through nb CLI's default redacted info output.",
"Register the generated stdio MCP configuration in the user's client.",
"Call the MCP tool crm_health and require ok=true.",
],
@@ -330,6 +539,11 @@ def build_parser() -> argparse.ArgumentParser:
set_env.add_argument("env_name")
set_env.add_argument("--json", action="store_true", dest="as_json")
connect_key = subparsers.add_parser("connect-key")
connect_key.add_argument("--api-base-url", required=True)
connect_key.add_argument("--env-name", default=DEFAULT_ENV_NAME)
connect_key.add_argument("--json", action="store_true", dest="as_json")
client_config = subparsers.add_parser("client-config")
client_config.add_argument("--enable-writes", action="store_true")
return parser
@@ -351,6 +565,14 @@ def main(argv: Sequence[str] | None = None) -> int:
payload = set_config_env(args.project_root, args.env_name)
_print_result(payload, as_json=args.as_json)
return 0
if args.command == "connect-key":
payload = connect_with_personal_key(
args.project_root,
args.api_base_url,
env_name=args.env_name,
)
_print_result(payload, as_json=args.as_json)
return 0
if args.command == "client-config":
print(json.dumps(render_client_config(args.project_root, enable_writes=args.enable_writes), indent=2))
return 0
scripts/manage_mcp_service.py
+1 -1
View File
@@ -469,7 +469,7 @@ def service_status(*, project_root: Path) -> str:
display_host = "127.0.0.1" if host in {"0.0.0.0", "::"} else host
lines.append(f"MCP URL: http://{display_host}:{port}{path}")
lines.append(f"nb binary: {nb_bin}")
lines.append(f"NocoBase CLI env: {nb_env} (authenticate with `nb env auth {nb_env}`)")
lines.append(f"NocoBase CLI env: {nb_env} (personal API Key managed by the Agent installer)")
uploads_enabled = values.get(ENABLE_UPLOADS_ENV) == "1"
lines.append(f"Attachment uploads: {'enabled' if uploads_enabled else 'disabled (safe default)'}")
canonical_configured = bool(values.get(API_BASE_URL_ENV) and values.get(API_TOKEN_ENV))