251 lines
8.2 KiB
Python
251 lines
8.2 KiB
Python
"""The sole public interface of this project: the local uvm CLI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from collections.abc import Sequence
|
|
from ipaddress import IPv4Address
|
|
|
|
from .app import Application, build_application
|
|
from .domain import VmSpec
|
|
from .errors import UvmError, ValidationError
|
|
from .firecracker.config import vcpu_count
|
|
from .validation import parse_cpu, parse_password, parse_ram, parse_username
|
|
|
|
|
|
def build_parser(application: Application) -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog=application.settings.app_name,
|
|
description="Tiny Firecracker microVM CLI for Ubuntu.",
|
|
)
|
|
parser.add_argument(
|
|
"--serve",
|
|
action="store_true",
|
|
help="launch the FastAPI management server instead of running a CLI command",
|
|
)
|
|
parser.add_argument(
|
|
"--host",
|
|
default="127.0.0.1",
|
|
help="server bind address used with --serve; defaults to 127.0.0.1",
|
|
)
|
|
parser.add_argument(
|
|
"--port",
|
|
type=_port_number,
|
|
default=8000,
|
|
help="server bind port used with --serve; defaults to 8000",
|
|
)
|
|
subcommands = parser.add_subparsers(dest="command")
|
|
|
|
install = subcommands.add_parser(
|
|
"install",
|
|
description=(
|
|
"Set UVM_FIRECRACKER_SHA256, UVM_KERNEL_SHA256, and UVM_ROOTFS_SHA256 "
|
|
"to trusted digests. UVM_ALLOW_UNVERIFIED_DOWNLOADS=1 is an explicit "
|
|
"development-only opt-out."
|
|
),
|
|
help="install dependencies and assets; SHA-256 values are required by default",
|
|
)
|
|
install.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="redownload Firecracker, kernel, and rootfs even when local assets already exist",
|
|
)
|
|
|
|
create = subcommands.add_parser("create", help="create and boot a microVM")
|
|
create.add_argument(
|
|
"--cpu",
|
|
default=str(application.settings.default_vcpu),
|
|
help="CPU capacity, e.g. 1, 2, 0.5 (fractional CPU is advisory)",
|
|
)
|
|
create.add_argument(
|
|
"--ram",
|
|
default=str(application.settings.default_ram_mib),
|
|
help="RAM, e.g. 512, 1G, 512M",
|
|
)
|
|
create.add_argument(
|
|
"--host-ip",
|
|
dest="host_ip",
|
|
help="guest IP to assign; otherwise uvm allocates one",
|
|
)
|
|
create.add_argument(
|
|
"--username",
|
|
default=application.settings.default_ssh_user,
|
|
help="existing guest account to configure; defaults to root",
|
|
)
|
|
create.add_argument(
|
|
"--password",
|
|
default="root",
|
|
help="guest login password; defaults to root",
|
|
)
|
|
|
|
subcommands.add_parser("list", help="list VMs")
|
|
|
|
ssh = subcommands.add_parser("ssh", help="SSH into a VM")
|
|
ssh.add_argument("vm", help="VM ID or guest IP")
|
|
ssh.add_argument("--user", help="override the username stored for the VM")
|
|
ssh.add_argument("--key")
|
|
ssh.add_argument(
|
|
"--insecure-host-key",
|
|
action="store_true",
|
|
help="disable SSH host-key verification for this connection",
|
|
)
|
|
|
|
stop = subcommands.add_parser("stop", help="stop a VM")
|
|
stop.add_argument("vm")
|
|
|
|
destroy = subcommands.add_parser("destroy", help="stop and remove a VM")
|
|
destroy.add_argument("vm")
|
|
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
try:
|
|
application = build_application()
|
|
parser = build_parser(application)
|
|
args = parser.parse_args(argv)
|
|
if args.serve:
|
|
if args.command is not None:
|
|
parser.error("--serve cannot be combined with a CLI command")
|
|
return _run_server(application, args)
|
|
if args.command is None:
|
|
parser.error("a command is required unless --serve is used")
|
|
return _run_command(application, args)
|
|
except UvmError as error:
|
|
print(f"uvm: error: {error}", file=sys.stderr)
|
|
return error.exit_code
|
|
except KeyboardInterrupt:
|
|
print("uvm: interrupted", file=sys.stderr)
|
|
return 130
|
|
|
|
|
|
def _run_server(application: Application, args: argparse.Namespace) -> int:
|
|
from .server import run_server
|
|
|
|
run_server(application, host=args.host, port=args.port)
|
|
return 0
|
|
|
|
|
|
def _run_command(application: Application, args: argparse.Namespace) -> int:
|
|
if args.command == "install":
|
|
firecracker, assets = application.installer.install(force_assets=args.force)
|
|
application.state_store.initialize()
|
|
print()
|
|
print("uvm installed.")
|
|
print(f" Firecracker: {firecracker}")
|
|
print(f" Kernel: {assets.kernel}")
|
|
print(f" Rootfs: {assets.rootfs}")
|
|
print()
|
|
print("Next:")
|
|
if application.settings.allow_unverified_downloads:
|
|
print(" Keep UVM_ALLOW_UNVERIFIED_DOWNLOADS=1 for later create commands.")
|
|
print(" Run your UVM command with: create --cpu 1 --ram 512")
|
|
return 0
|
|
|
|
if args.command == "create":
|
|
guest_ip = _parse_guest_ip(args.host_ip)
|
|
vm = application.lifecycle.create(
|
|
VmSpec(
|
|
cpu=parse_cpu(args.cpu),
|
|
ram_mib=parse_ram(args.ram),
|
|
guest_ip=guest_ip,
|
|
username=parse_username(args.username),
|
|
password=parse_password(args.password),
|
|
)
|
|
)
|
|
if vm.cpu < 1.0:
|
|
print(
|
|
f"NOTE: requested {vm.cpu} CPU. Firecracker uses {vcpu_count(vm.cpu)} vCPU;"
|
|
)
|
|
print(" fractional CPU enforcement is not yet applied by this MVP.")
|
|
print()
|
|
print(f"VM created: {vm.id}")
|
|
print(f" IP: {vm.guest_ip}")
|
|
print(f" RAM: {vm.ram_mib} MiB")
|
|
print(f" CPU: {vm.cpu}")
|
|
print(f" TAP: {vm.tap}")
|
|
print(f" Username: {vm.username}")
|
|
print(f" SSH: ssh {vm.username}@{vm.guest_ip}")
|
|
print()
|
|
if vm.password == "root":
|
|
print("WARNING: the guest is using the default password 'root'. Change it promptly.")
|
|
return 0
|
|
|
|
if args.command == "list":
|
|
vms = application.lifecycle.list_vms()
|
|
if not vms:
|
|
print("No VMs.")
|
|
return 0
|
|
print(
|
|
f"{'ID':<37} {'IP':<16} {'USER':<16} {'CPU':<7} "
|
|
f"{'RAM':<8} {'STATUS':<10} PID"
|
|
)
|
|
for listed in vms:
|
|
vm = listed.vm
|
|
pid = vm.pid if vm.pid is not None else ""
|
|
print(
|
|
f"{vm.id:<37} {vm.guest_ip:<16} {vm.username:<16} {vm.cpu:<7} "
|
|
f"{vm.ram_mib:<8} {listed.observed_status:<10} {pid}"
|
|
)
|
|
return 0
|
|
|
|
if args.command == "ssh":
|
|
vm = application.lifecycle.find_for_ssh(args.vm)
|
|
command = ["ssh"]
|
|
if args.key:
|
|
command.extend(("-i", args.key))
|
|
if args.insecure_host_key:
|
|
command.extend(
|
|
(
|
|
"-o",
|
|
"StrictHostKeyChecking=no",
|
|
"-o",
|
|
"UserKnownHostsFile=/dev/null",
|
|
)
|
|
)
|
|
else:
|
|
command.extend(
|
|
(
|
|
"-o",
|
|
f"HostKeyAlias=uvm-{vm.id}",
|
|
"-o",
|
|
"StrictHostKeyChecking=accept-new",
|
|
)
|
|
)
|
|
command.append(f"{args.user or vm.username}@{vm.guest_ip}")
|
|
os.execvp(command[0], command)
|
|
return 0
|
|
|
|
if args.command == "stop":
|
|
vm = application.lifecycle.stop(args.vm)
|
|
print(f"Stopped {vm.id}")
|
|
return 0
|
|
|
|
if args.command == "destroy":
|
|
vm = application.lifecycle.destroy(args.vm)
|
|
print(f"Destroyed {vm.id}")
|
|
return 0
|
|
|
|
raise UvmError(f"unsupported command: {args.command}")
|
|
|
|
|
|
def _parse_guest_ip(value: str | None) -> IPv4Address | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return IPv4Address(value)
|
|
except ValueError as error:
|
|
raise ValidationError(f"invalid guest IP: {value}") from error
|
|
|
|
|
|
def _port_number(value: str) -> int:
|
|
try:
|
|
port = int(value)
|
|
except ValueError as error:
|
|
raise argparse.ArgumentTypeError("port must be an integer") from error
|
|
if not 1 <= port <= 65535:
|
|
raise argparse.ArgumentTypeError("port must be between 1 and 65535")
|
|
return port
|