super/box
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cli): add 'box ssh' command suite (mint, list, show) with auto signers registration

Browse Source
This commit is contained in:
operator
2026-10-05 15:33:38 +00:00
parent 26535791e5
commit 7f92679281
2 changed files with 222 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/box-ctl.py
+132
View File
@@ -1382,6 +1382,112 @@ def act_loop_breaks():
fail("LOOP_ERROR", str(e))
# ---------------------------------------------------------------------------
# SSH key minting & management
# ---------------------------------------------------------------------------
KEY_DIR = Path("/home/super/.ssh")
ALLOWED_SIGNERS_PATHS = [
NETVM_ROOT / "dm-signers" / "allowed_signers",
Path("/home/super/.exec-signers")
]
def act_ssh_mint(name, force=False):
"""Mint a new ed25519 SSH keypair and register it in allowed_signers."""
key_name = f"id_{name}"
priv_path = KEY_DIR / key_name
pub_path = KEY_DIR / f"{key_name}.pub"
if priv_path.exists() and not force:
fail("KEY_EXISTS", f"SSH key {priv_path} already exists -- pass --force to overwrite",
{"key_path": str(priv_path)})
if priv_path.exists() and force:
try:
priv_path.unlink()
if pub_path.exists():
pub_path.unlink()
except Exception as e:
fail("IO_ERROR", f"Failed removing old key: {e}")
cmd = ["ssh-keygen", "-t", "ed25519", "-N", "", "-C", f"{name}@netvm", "-f", str(priv_path)]
r = subprocess.run(cmd, capture_output=True, text=True)
if r.returncode != 0:
fail("KEYGEN_FAILED", f"ssh-keygen failed: {r.stderr.strip()}")
pub_line = pub_path.read_text(encoding="utf-8").strip()
pub_key_body = pub_line.split()[1] if len(pub_line.split()) > 1 else pub_line
# Save to dm-signers/{name}.pub
ds_pub = NETVM_ROOT / "dm-signers" / f"{name}.pub"
try:
ds_pub.write_text(pub_line + "\n", encoding="utf-8")
except Exception:
pass
# Register into signers files
registered_files = []
for sp in ALLOWED_SIGNERS_PATHS:
try:
lines = sp.read_text(encoding="utf-8").splitlines() if sp.exists() else []
new_lines = []
found = False
for line in lines:
if not line.strip():
continue
parts = line.split()
if parts[0] in (name, f"operator-{name}"):
new_lines.append(f"{parts[0]} ssh-ed25519 {pub_key_body} {name}@netvm")
found = True
else:
new_lines.append(line)
if not found:
new_lines.append(f"{name} ssh-ed25519 {pub_key_body} {name}@netvm")
new_lines.append(f"operator-{name} ssh-ed25519 {pub_key_body} {name}@netvm")
sp.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
registered_files.append(str(sp))
except Exception as e:
sys.stderr.write(f"Warning: failed updating {sp}: {e}\n")
audit("ssh-mint", f"name={name} priv={priv_path}")
out(True, name=name, private_key=str(priv_path), public_key=str(pub_path),
public_key_content=pub_line, registered_in=registered_files)
def act_ssh_list():
"""List managed SSH keypairs and registrations."""
keys = []
if KEY_DIR.exists():
for p in sorted(KEY_DIR.glob("id_*")):
if not p.name.endswith(".pub"):
pub_p = p.with_name(p.name + ".pub")
pub_content = pub_p.read_text(encoding="utf-8").strip() if pub_p.exists() else None
keys.append({
"name": p.name[3:],
"private_key": str(p),
"has_public": pub_p.exists(),
"public_key": pub_content,
})
out(True, keys=keys, count=len(keys))
def act_ssh_show(name):
"""Show public key details and fingerprints for an identity."""
priv_path = KEY_DIR / f"id_{name}"
pub_path = KEY_DIR / f"id_{name}.pub"
if not pub_path.exists():
fail("NOT_FOUND", f"No public key found for {name} at {pub_path}")
pub_line = pub_path.read_text(encoding="utf-8").strip()
cmd = ["ssh-keygen", "-lf", str(pub_path)]
r = subprocess.run(cmd, capture_output=True, text=True)
fp = r.stdout.strip() if r.returncode == 0 else ""
out(True, name=name, private_key=str(priv_path), public_key=str(pub_path),
public_key_content=pub_line, fingerprint=fp)
def act_loop_resolve(dm_id, note=None):
f_path = NETVM_ROOT / "followups.json"
resolved = False
@@ -2845,6 +2951,32 @@ def main(argv):
fail("BAD_NAME", "usage: loop-resolve <dm_id> [note]")
note = rest[1] if len(rest) > 1 else None
act_loop_resolve(rest[0], note=note)
elif action in ("ssh", "ssh-mint", "ssh-list", "ssh-show"):
if action == "ssh-mint":
if not rest:
fail("BAD_ARGS", "usage: ssh-mint <name> [--force]")
act_ssh_mint(rest[0], force=("--force" in rest[1:]))
elif action == "ssh-list":
act_ssh_list()
elif action == "ssh-show":
if not rest:
fail("BAD_ARGS", "usage: ssh-show <name>")
act_ssh_show(rest[0])
elif action == "ssh":
if not rest or rest[0] not in ("mint", "list", "show"):
fail("BAD_NAME", "usage: ssh mint|list|show [...]")
sub = rest[0]
args = rest[1:]
if sub == "mint":
if not args:
fail("BAD_ARGS", "usage: ssh mint <name> [--force]")
act_ssh_mint(args[0], force=("--force" in args[1:]))
elif sub == "list":
act_ssh_list()
elif sub == "show":
if not args:
fail("BAD_ARGS", "usage: ssh show <name>")
act_ssh_show(args[0])
elif action == "loop-remediate":
dry = "--dry-run" in rest
act_loop_remediate(dry_run=dry)
bin/super-cli.py
+90
View File
@@ -3042,6 +3042,73 @@ def cmd_loop_remediate(args):
# ---------------------------------------------------------------------------
# CLI Argument Parser Setup
# ---------------------------------------------------------------------------
def cmd_ssh_mint(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "ssh-mint", args.name]
if args.force:
cmd.append("--force")
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_green(f"✓ Minted ed25519 SSH keypair for '{args.name}'"))
print(f" Private key: {d.get('private_key')}")
print(f" Public key: {d.get('public_key')}")
print(f" Public key text: {d.get('public_key_content')}")
print(c_dim(" Registered into: ") + ", ".join(d.get("registered_in", [])))
else:
print(c_red(f"✗ Failed: {d.get('error')} ({d.get('code')})"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_ssh_list(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "ssh-list"]
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
keys = d.get("keys", [])
print("\n" + c_bold(f"=== MANAGED SSH KEYS ({len(keys)}) ===") + "\n")
headers = ["NAME", "PRIVATE KEY", "PUB STATUS", "PUBLIC KEY PREVIEW"]
rows = []
for k in keys:
pub_txt = k.get("public_key") or ""
preview = (pub_txt[:32] + "..." + pub_txt[-16:]) if len(pub_txt) > 48 else pub_txt
rows.append([c_cyan(k.get("name")), k.get("private_key"), c_green("✓ present") if k.get("has_public") else c_red("missing"), preview])
print_table(headers, rows)
print("\n" + c_dim(" Mint new key: box ssh mint <name> [--force]") + "\n")
else:
print(c_red(f"✗ Failed: {d.get('error')}"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_ssh_show(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "ssh-show", args.name]
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_bold(f"=== SSH KEY DETAILS: {args.name} ==="))
print(f" Private key: {d.get('private_key')}")
print(f" Public key: {d.get('public_key')}")
print(f" Fingerprint: {d.get('fingerprint')}")
print(f"\n Public Key:\n {c_cyan(d.get('public_key_content'))}\n")
else:
print(c_red(f"✗ Failed: {d.get('error')}"))
except Exception:
sys.stdout.write(res.stdout)
def build_parser():
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--json", action="store_true", help="Output machine-readable JSON")
@@ -3233,6 +3300,19 @@ def build_parser():
cred_sub.add_parser("list", parents=[common], help="List all registered nodes and vitality statuses")
# Domain: SSH
p_ssh = subparsers.add_parser("ssh", parents=[common], help="Mint and manage fleet SSH signing/access keys")
ssh_sub = p_ssh.add_subparsers(dest="action")
p_s_mint = ssh_sub.add_parser("mint", parents=[common], help="Mint a new ed25519 SSH keypair and register in allowed_signers")
p_s_mint.add_argument("name", help="Identity/agent name (e.g. 646, pip, dev, canary)")
p_s_mint.add_argument("--force", action="store_true", help="Overwrite existing keypair")
p_s_list = ssh_sub.add_parser("list", parents=[common], help="List managed SSH keypairs and registration status")
p_s_show = ssh_sub.add_parser("show", parents=[common], help="Show public key content and fingerprint")
p_s_show.add_argument("name", help="Identity/agent name")
# Domain: HARVEST
p_harvest = subparsers.add_parser("harvest", parents=[common], help="Readback & response harvesting engine")
harvest_sub = p_harvest.add_subparsers(dest="action")
@@ -3539,6 +3619,16 @@ def main():
cmd_cred_meta_audit(args)
else:
parser.print_help()
elif args.domain == "ssh":
act = getattr(args, "action", None)
if not act or act == "list":
cmd_ssh_list(args)
elif act == "mint":
cmd_ssh_mint(args)
elif act == "show":
cmd_ssh_show(args)
else:
p_ssh.print_help()
elif args.domain == "harvest":
act = getattr(args, "action", None)
if not act or act == "status":