118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import getpass
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from collections.abc import Sequence
|
|
from dataclasses import replace
|
|
|
|
from host_agent.app import create_application
|
|
from host_agent.config import load_host_agent_config
|
|
from host_agent.instance_lock import InstanceAlreadyRunningError
|
|
from host_agent.local_account import LocalAccountStore
|
|
from host_agent.mcp_token import McpTokenStore
|
|
|
|
|
|
class LocalAccountSetupError(RuntimeError):
|
|
"""Raised when local account bootstrap cannot proceed."""
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> None:
|
|
_load_dotenv()
|
|
parser = argparse.ArgumentParser(description="Run the Device Host Agent")
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
subparsers.add_parser("setup", help="Create the local operator account")
|
|
subparsers.add_parser(
|
|
"mcp-token",
|
|
help="Print the MCP server bearer token (generating if missing)",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.command == "mcp-token":
|
|
_print_mcp_token()
|
|
return
|
|
|
|
try:
|
|
if args.command == "setup":
|
|
_run_setup()
|
|
return
|
|
config = _resolve_config_with_local_account()
|
|
except LocalAccountSetupError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
raise SystemExit(1) from exc
|
|
|
|
try:
|
|
create_application(config=config).run()
|
|
except InstanceAlreadyRunningError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
raise SystemExit(1) from exc
|
|
|
|
|
|
def _run_setup() -> None:
|
|
config = load_host_agent_config()
|
|
store = LocalAccountStore(config.local_account_path)
|
|
if store.load() is not None:
|
|
confirm = (
|
|
input("A local account already exists. Overwrite it? [y/N] ")
|
|
.strip()
|
|
.lower()
|
|
)
|
|
if confirm != "y":
|
|
print("Setup cancelled; existing account left unchanged.")
|
|
return
|
|
account = _prompt_and_create(store)
|
|
print(f"Local account '{account.username}' created.")
|
|
|
|
|
|
def _print_mcp_token() -> None:
|
|
config = load_host_agent_config()
|
|
store = McpTokenStore(config.identity_path.parent / "host_mcp_token.json")
|
|
print(store.load_or_create().token)
|
|
|
|
|
|
def _resolve_config_with_local_account():
|
|
config = load_host_agent_config()
|
|
store = LocalAccountStore(config.local_account_path)
|
|
account = store.load()
|
|
if account is None:
|
|
if not sys.stdin.isatty():
|
|
raise LocalAccountSetupError(
|
|
"no local account configured; run `device-host-agent setup` "
|
|
"on an interactive terminal to create one"
|
|
)
|
|
account = _prompt_and_create(store)
|
|
if not config.display_name:
|
|
config = replace(config, display_name=account.username)
|
|
return config
|
|
|
|
|
|
def _prompt_and_create(store: LocalAccountStore):
|
|
username = input("Username: ").strip()
|
|
if not username:
|
|
raise LocalAccountSetupError("username must not be empty")
|
|
password = getpass.getpass("Password: ")
|
|
confirm = getpass.getpass("Confirm password: ")
|
|
if not password:
|
|
raise LocalAccountSetupError("password must not be empty")
|
|
if password != confirm:
|
|
raise LocalAccountSetupError("passwords do not match")
|
|
return store.create(username, password)
|
|
|
|
|
|
def _load_dotenv() -> None:
|
|
"""Load a simple repository-root .env without overriding shell values."""
|
|
path = Path.cwd() / ".env"
|
|
if not path.is_file():
|
|
return
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
name, value = line.split("=", 1)
|
|
name = name.strip()
|
|
value = value.strip()
|
|
if name and name not in os.environ:
|
|
os.environ[name] = value
|