#!/usr/bin/python3
# terractl-remote - manage Terraria servers on other hosts (terraria-server-manager)
#
# Run by terractl (never directly) for:
#   terractl daemon ...             the daemon other hosts connect to (terractld.service)
#   terractl remote ...             the hosts this user manages ("remotes")
#   terractl --host <remote> ...    any terractl command, run on a remote
#   terractl copy|move ...          instances between hosts, or within one
#
# How it fits together: the daemon is a small HTTPS server that does everything
# by running terractl on its own host, so paths, checks and file ownership are
# exactly as if someone had run terractl there. Copy and move are 'export' on
# the source, a file transfer, and 'import' on the destination; import adapts
# the instance to the destination's folders and free ports.
#
# Security: the daemon's certificate is self-signed. A computer joins with a
# one-time token from 'terractl daemon enable' or 'daemon token', which holds the certificate's
# fingerprint and a secret. The client checks that fingerprint on every
# connection, so nothing in between can pose as the daemon, and trades the
# secret for a random key of its own, sent with each request. The daemon stores
# only SHA-256 hashes of keys and secrets. A trusted computer can do what
# 'sudo terractl' can do on that host, except change the daemon itself.
#
# Python standard library only (3.9 or newer).

import base64
import codecs
import errno
import fcntl
import hashlib
import hmac
import http.client
import http.server
import json
import os
import pwd
import re
import secrets
import select
import shlex
import shutil
import signal
import socket
import socketserver
import ssl
import stat
import subprocess
import sys
import tarfile
import tempfile
import threading
import time
import traceback
import urllib.parse
from pathlib import PurePosixPath

API = 1                          # changes only if old and new versions can't talk
DEFAULT_PORT = 7776              # one below the first game port terractl hands out
DAEMON_UNIT = "terractld.service"
TOKEN_PREFIX = "tctl1_"
TOKEN_HOURS = 24
TRANSFER_HOURS = 6               # files sent to the daemon but never used are removed after this
CHUNK = 1024 * 1024
PING_SECONDS = 20                # a quiet stream sends a ping this often...
STREAM_TIMEOUT = 90              # ...so hearing nothing for this long means the connection is gone
IO_TIMEOUT = 120
MAX_CONNECTIONS = 64

TERRACTL = os.environ.get("TERRACTL_BIN") or "terractl"
VERSION = os.environ.get("TERRACTL_VERSION") or "development"
STATE_DIR = os.environ.get("TERRACTL_STATE_DIR") or "/var/lib/terractl"
CERT, KEY, TRUST = "server.crt", "server.key", "trust.json"

NAME_RE = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]*")         # instances and remotes (valid_name)
CLIENT_RE = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.@-]{0,63}")
ZIP_RE = re.compile(r"[A-Za-z0-9_][A-Za-z0-9._+-]*\.zip")
OUTPUT_RE = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]*\.(tar\.gz|tgz)")
VIA_RE = re.compile(r"--host [A-Za-z0-9_.:\[\]%-]{1,200}")


class Fail(Exception):
    """Shown to the user as 'terractl: <message>'; exit status 1."""


class Unreachable(Fail):
    """Couldn't connect to a daemon, or it wasn't the expected one."""


def die(message):
    raise Fail(message)


def log(message):
    print(f"terractl: {message}", file=sys.stderr, flush=True)


def human_size(n):
    if n < 1000:
        return f"{n} bytes"
    for unit in ("KB", "MB", "GB", "TB"):
        n /= 1000
        if n < 1000 or unit == "TB":
            return f"{n:.1f} {unit}"


def sha256(text):
    return hashlib.sha256(text.encode()).hexdigest()


def split_hostport(text, default_port=DEFAULT_PORT):
    """'host', 'host:port', '[v6addr]:port', 'v6addr' or 'port' -> (host, port).
    An empty host means all addresses."""
    text = text.strip()
    host, port = text, ""
    if text.startswith("["):
        host, sep, rest = text[1:].partition("]")
        if not sep or (rest and not rest.startswith(":")):
            die(f"invalid address '{text}'")
        port = rest[1:]
    elif text.isdigit():
        host, port = "", text
    elif text.count(":") == 1:
        host, port = text.split(":")
    if port == "":
        return host, default_port
    if not port.isdigit() or not 0 < int(port) < 65536:
        die(f"invalid port in '{text}'")
    return host, int(port)


def join_hostport(host, port):
    return f"[{host}]:{port}" if ":" in host else f"{host}:{port}"


def describe_listen(host, port):
    if host in ("", "::"):
        return f"all addresses, port {port}"
    if host == "0.0.0.0":
        return f"all IPv4 addresses, port {port}"
    return join_hostport(host, port)


def listen_setting():
    """Where the daemon listens: TERRACTL_LISTEN from /etc/default/terractl."""
    return split_hostport(os.environ.get("TERRACTL_LISTEN", ""))


def parse_info(text):
    """'terractl info' output -> {key: [values]} (some keys repeat)."""
    info = {}
    for line in text.splitlines():
        key, sep, value = line.partition("=")
        if sep:
            info.setdefault(key, []).append(value)
    return info


def one(info, key, default=""):
    values = info.get(key)
    return values[-1] if values else default


def terractl(args, capture=False, show_stdout=True):
    """Runs terractl on this host. Its messages go straight to the terminal."""
    if capture:
        stdout = subprocess.PIPE
    else:
        stdout = None if show_stdout else subprocess.DEVNULL
    try:
        return subprocess.run([TERRACTL, *args], stdout=stdout, text=True)
    except OSError as e:
        die(f"can't run {TERRACTL}: {e.strerror}")


def systemctl(*args):
    try:
        return subprocess.run(["systemctl", *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    except OSError:
        return subprocess.CompletedProcess(args, 1, "", "systemctl not found")


def big_tempdir():
    """A private folder for archives, on disk rather than in memory (/tmp may be tmpfs)."""
    try:
        return tempfile.mkdtemp(prefix="terractl-", dir="/var/tmp")
    except OSError:
        return tempfile.mkdtemp(prefix="terractl-")


# ---------------------------------------------------------- daemon state ----
# <STATE_DIR>/server.crt, server.key  the daemon's certificate
#            /trust.json              trusted computers and waiting tokens (hashed)
#            /transfers/              files in transit (cleared when the daemon starts)

def state_path(*parts):
    return os.path.join(STATE_DIR, *parts)


def need_root(cmdline):
    if os.geteuid() != 0:
        die(f"this needs root: sudo terractl {cmdline}")


def open_state(cmdline):
    try:
        os.makedirs(STATE_DIR, mode=0o755, exist_ok=True)
    except PermissionError:
        die(f"this needs root: sudo terractl {cmdline}")
    if not os.access(STATE_DIR, os.W_OK):
        die(f"this needs root: sudo terractl {cmdline}")


class StateLock:
    """Serialises changes to trust.json between the daemon and 'terractl daemon token/revoke'."""

    def __enter__(self):
        self.fd = os.open(state_path(".lock"), os.O_RDWR | os.O_CREAT, 0o600)
        fcntl.flock(self.fd, fcntl.LOCK_EX)

    def __exit__(self, *exc):
        os.close(self.fd)


def ensure_cert():
    crt, key = state_path(CERT), state_path(KEY)
    if os.path.exists(crt) and os.path.exists(key):
        return
    with StateLock():
        if os.path.exists(crt) and os.path.exists(key):
            return
        host = socket.gethostname() or "localhost"
        new_crt, new_key = crt + ".new", key + ".new"
        old_umask = os.umask(0o077)
        try:
            proc = subprocess.run(
                ["openssl", "req", "-x509", "-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:prime256v1",
                 "-sha256", "-nodes", "-days", "3650", "-subj", f"/CN=terractld@{host}",
                 "-keyout", new_key, "-out", new_crt],
                stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
        except OSError:
            manager = "dnf" if shutil.which("dnf") and not shutil.which("apt-get") else "apt"
            die(f"openssl is not installed (sudo {manager} install openssl)")
        finally:
            os.umask(old_umask)
        if proc.returncode != 0:
            die(f"creating the daemon's certificate failed: {proc.stderr.strip()}")
        os.chmod(new_crt, 0o644)
        os.replace(new_key, key)
        os.replace(new_crt, crt)


def cert_fingerprint():
    with open(state_path(CERT)) as f:
        return hashlib.sha256(ssl.PEM_cert_to_DER_cert(f.read())).hexdigest()


def load_trust():
    try:
        with open(state_path(TRUST)) as f:
            trust = json.load(f)
    except FileNotFoundError:
        trust = {}
    except ValueError as e:
        die(f"{state_path(TRUST)} is damaged ({e})")
    trust.setdefault("clients", [])
    trust.setdefault("tokens", [])
    return trust


def save_trust(trust):
    tmp = state_path(TRUST + ".new")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(trust, f, indent=2)
        f.write("\n")
    os.replace(tmp, state_path(TRUST))


def make_token(fields):
    raw = json.dumps(fields, separators=(",", ":")).encode()
    return TOKEN_PREFIX + base64.urlsafe_b64encode(raw).decode().rstrip("=")


def read_token(text):
    text = text.strip()
    if not text.startswith(TOKEN_PREFIX):
        die(f"that isn't a terractl token (they start with {TOKEN_PREFIX})")
    body = text[len(TOKEN_PREFIX):]
    try:
        fields = json.loads(base64.urlsafe_b64decode(body + "=" * (-len(body) % 4)))
    except ValueError:
        fields = None
    if (not isinstance(fields, dict) or not isinstance(fields.get("s"), str)
            or not re.fullmatch(r"[0-9a-f]{64}", str(fields.get("fp", "")))):
        die("the token is damaged; copy it again (it's one long line)")
    return fields


def host_addresses(listen_host, port):
    """Addresses to put in a token, likeliest to be reachable first."""
    if listen_host not in ("", "::", "0.0.0.0"):
        return [join_hostport(listen_host, port)]
    found = []
    try:
        # The address of the default route; nothing is actually sent
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.connect(("192.0.2.1", 9))
            found.append(s.getsockname()[0])
    except OSError:
        pass
    try:
        found += subprocess.run(["hostname", "-I"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                                text=True, timeout=5).stdout.split()
    except (OSError, subprocess.SubprocessError):
        pass
    result = []
    for addr in found:
        if addr.startswith(("127.", "fe80:")) or addr == "::1" or addr in result:
            continue
        if listen_host == "0.0.0.0" and ":" in addr:
            continue
        result.append(addr)
    return [join_hostport(a, port) for a in result]


# --------------------------------------------------------------- daemon ----

EXEC_COMMANDS = {"list", "info", "new", "start", "stop", "restart", "enable", "disable", "cmd", "watch",
                 "mods", "reinstall", "upgrade", "fix-perms", "doctor", "backup", "fetch", "delete", "rename", "export", "import",
                 "-V", "--version", "version", "-h", "--help", "help"}


class HttpError(Exception):
    def __init__(self, status, message):
        super().__init__(message)
        self.status = status


def dlog(message):
    """The daemon's log (the journal, under systemd)."""
    print(message, file=sys.stderr, flush=True)


def child_env(via=""):
    """Environment for terractl run by the daemon. Minimal, and in particular
    without systemd's INVOCATION_ID, which 'terractl stop' would take to mean
    it's being run as terraria@.service's ExecStop."""
    env = {"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
           "LANG": "C.UTF-8", "HOME": os.environ.get("HOME", "/root"), "TERM": "dumb",
           "SYSTEMD_COLORS": "0", "SYSTEMD_PAGER": "", "PAGER": "cat"}
    env.update((k, v) for k, v in os.environ.items() if k.startswith("TERRARIA_"))
    if via:
        env["TERRACTL_VIA"] = via
    return env


def local_host_info(env=None):
    proc = subprocess.run([TERRACTL, "info"], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          text=True, cwd="/", env=env)
    if proc.returncode != 0:
        raise RuntimeError(f"terractl info failed: {proc.stderr.strip()}")
    return parse_info(proc.stdout)


class Transfers:
    """Files sent to the daemon, and exports waiting to be fetched. Each lives in
    its own folder under <state>/transfers and is used once."""

    def __init__(self, root):
        self.root = root
        self.lock = threading.Lock()
        self.items = {}
        shutil.rmtree(root, ignore_errors=True)      # left over from before a restart
        os.makedirs(root, mode=0o700)

    def new_dir(self):
        return tempfile.mkdtemp(dir=self.root)

    def add(self, path):
        tid = secrets.token_hex(16)
        with self.lock:
            self.items[tid] = {"path": path, "dir": os.path.dirname(path), "time": time.time()}
        return tid

    def take(self, tid):
        with self.lock:
            item = self.items.pop(tid, None) if isinstance(tid, str) else None
        if item is None:
            raise HttpError(404, f"that file is gone (unused files are removed after {TRANSFER_HOURS} hours)")
        return item

    @staticmethod
    def remove(item):
        shutil.rmtree(item["dir"], ignore_errors=True)

    def expire(self):
        cutoff = time.time() - TRANSFER_HOURS * 3600
        with self.lock:
            old = [tid for tid, item in self.items.items() if item["time"] < cutoff]
            gone = [self.items.pop(tid) for tid in old]
        for item in gone:
            self.remove(item)


def extract_upload(tar_path, dest):
    """Unpacks the files and folders sent for 'mods add'. Only plain files and
    folders are accepted, and nothing may land outside dest."""
    try:
        with tarfile.open(tar_path) as tar:
            for member in tar:
                parts = PurePosixPath(member.name).parts
                if not parts:
                    continue
                if parts[0] == "/" or ".." in parts:
                    raise HttpError(400, f"unsafe path in the files sent: {member.name}")
                target = os.path.join(dest, *parts)
                if member.isdir():
                    os.makedirs(target, mode=0o755, exist_ok=True)
                elif member.isfile():
                    os.makedirs(os.path.dirname(target), mode=0o755, exist_ok=True)
                    with tar.extractfile(member) as src, open(target, "xb") as out:
                        shutil.copyfileobj(src, out, CHUNK)
                    os.chmod(target, 0o644)
                else:
                    raise HttpError(400, f"the files sent include a link or special file: {member.name}")
    except (tarfile.TarError, FileExistsError) as e:
        raise HttpError(400, f"the files sent aren't a valid archive ({e})")


def unique_client_name(wanted, clients):
    base = wanted[:60] if isinstance(wanted, str) and CLIENT_RE.fullmatch(wanted[:60]) else "computer"
    taken = {c.get("name") for c in clients}
    name, n = base, 2
    while name in taken:
        name, n = f"{base}-{n}", n + 1
    return name


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = "terractld"
    sys_version = ""
    timeout = IO_TIMEOUT

    def log_message(self, format, *args):
        pass       # requests are logged by what they do

    def do_GET(self):
        self.dispatch()

    do_POST = do_PUT = do_DELETE = do_GET

    def dispatch(self):
        self.who = None
        self.started = self.streaming = self.body_read = False
        try:
            if "Transfer-Encoding" in self.headers:
                raise HttpError(411, "send a Content-Length")
            path = urllib.parse.urlsplit(self.path).path
            parts = [urllib.parse.unquote(p) for p in path.strip("/").split("/")]
            if parts[0] != "1.0":
                raise HttpError(404, "not found")
            self.route(self.command, parts[1:])
        except HttpError as e:
            self.fail(e.status, str(e))
        except (ConnectionError, TimeoutError, ssl.SSLError):
            self.close_connection = True
        except OSError as e:
            self.fail(500, e.strerror or str(e))
        except Exception:
            dlog(f"{self.who or self.client_address[0]}: {self.command} {self.path} failed:\n"
                 f"{traceback.format_exc()}")
            self.fail(500, "internal error; see the daemon's log: journalctl -u terractld")

    def route(self, method, p):
        if method == "POST" and p == ["trust", "join"]:
            return self.join()
        self.who = self.authenticate()
        if method == "GET" and p == []:
            return self.send_json(200, self.server.host_info())
        if method == "POST" and p == ["exec"]:
            return self.run_command()
        if method == "PUT" and p == ["files"]:
            return self.put_file()
        if len(p) == 2 and p[0] == "files" and method in ("GET", "DELETE"):
            item = self.server.transfers.take(p[1])
            try:
                if method == "GET":
                    self.send_file(item["path"])
                else:
                    self.send_json(200, {})
            finally:
                self.server.transfers.remove(item)
            return
        if len(p) == 2 and p[0] == "console" and method == "POST":
            return self.console(p[1])
        if len(p) == 2 and p[0] == "dist" and method in ("GET", "PUT"):
            return self.get_zip(p[1]) if method == "GET" else self.put_zip(p[1])
        raise HttpError(404, "not found")

    # ---- plumbing

    def authenticate(self):
        auth = self.headers.get("Authorization", "")
        if auth.startswith("Bearer "):
            digest = sha256(auth[7:].strip())
            for client in load_trust()["clients"]:
                if hmac.compare_digest(client.get("key_sha256", ""), digest):
                    return client["name"]
        dlog(f"{self.client_address[0]}: refused (not a trusted key)")
        raise HttpError(401, "not a trusted key")

    def fail(self, status, message):
        self.close_connection = True
        try:
            if self.streaming:
                self.frame({"error": message})
            elif not self.started:
                # Closing with unread data would reset the connection before the
                # client reads the error; small bodies are read first (big ones
                # are uploads, which clients only start once they know they're in)
                length = self.content_length() if not self.body_read else 0
                if length <= 1 << 20:
                    self.rfile.read(length)
                self.send_json(status, {"error": message}, close=True)
        except (OSError, HttpError):
            pass

    def send_json(self, status, obj, close=False):
        body = (json.dumps(obj) + "\n").encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        if close:
            self.send_header("Connection", "close")
            self.close_connection = True
        self.end_headers()
        self.started = True
        self.wfile.write(body)

    def content_length(self):
        try:
            length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            length = -1
        if length < 0:
            raise HttpError(400, "bad Content-Length")
        return length

    def read_json(self):
        length = self.content_length()
        if length > 1 << 20:
            raise HttpError(413, "request too large")
        self.body_read = True
        try:
            data = json.loads(self.rfile.read(length) or b"{}")
        except ValueError:
            data = None
        if not isinstance(data, dict):
            raise HttpError(400, "bad request")
        return data

    def receive(self, path, length):
        """Saves the request body to a new file."""
        self.body_read = True
        with open(path, "xb") as f:
            left = length
            while left:
                data = self.rfile.read(min(CHUNK, left))
                if not data:
                    raise HttpError(400, "the upload was cut off")
                f.write(data)
                left -= len(data)

    def send_file(self, path):
        with open(path, "rb") as f:
            size = os.fstat(f.fileno()).st_size
            self.send_response(200)
            self.send_header("Content-Type", "application/octet-stream")
            self.send_header("Content-Length", str(size))
            self.end_headers()
            self.started = True
            shutil.copyfileobj(f, self.wfile, CHUNK)

    def frame(self, obj):
        self.wfile.write((json.dumps(obj) + "\n").encode())

    # ---- endpoints

    def join(self):
        req = self.read_json()
        secret, wanted, old_key = req.get("secret"), req.get("name"), req.get("old_key")
        if not isinstance(secret, str):
            raise HttpError(400, "bad request")
        digest = sha256(secret)
        with StateLock():
            trust = load_trust()
            now = time.time()
            trust["tokens"] = [t for t in trust["tokens"] if t.get("expires", 0) > now]
            token = next((t for t in trust["tokens"]
                          if hmac.compare_digest(t.get("secret_sha256", ""), digest)), None)
            if token is None:
                save_trust(trust)
                dlog(f"{self.client_address[0]}: refused a join (unknown, used or expired token)")
                raise HttpError(403, "this token is not valid (it may have been used already, or "
                                     "expired); create a new one there: sudo terractl daemon token")
            trust["tokens"].remove(token)
            key = secrets.token_urlsafe(32)
            # A computer that's trusted already (it sends its old key) keeps its
            # entry and name; a new one is named after its hostname
            old = None
            if isinstance(old_key, str):
                old = next((c for c in trust["clients"]
                            if hmac.compare_digest(c.get("key_sha256", ""), sha256(old_key))), None)
            if old:
                name = old["name"]
                trust["clients"].remove(old)
            else:
                name = unique_client_name(wanted, trust["clients"])
            trust["clients"].append({"name": name, "key_sha256": sha256(key),
                                     "added": time.strftime("%Y-%m-%d %H:%M"),
                                     "from": self.client_address[0]})
            save_trust(trust)
        dlog(f"{self.client_address[0]}: joined as '{name}'")
        self.send_json(200, {"name": name, "key": key})

    def resolve(self, args, cleanup):
        """The command line to run from a client's arguments. Where terractl takes
        file paths, only files sent to the daemon are accepted, never paths on
        this host: import <name> <file>, export <name> [file], mods <name> add
        <files or folders>."""
        if not isinstance(args, list) or not args or not all(isinstance(a, (str, dict)) for a in args):
            raise HttpError(400, "bad request")
        command = args[0]
        if not isinstance(command, str) or command not in EXEC_COMMANDS:
            raise HttpError(400, f"'{command}' can't be run from another host")
        mods_add = command == "mods" and len(args) > 2 and args[2] == "add"
        argv, shown, output = [command], [command], None
        for i, arg in enumerate(args[1:], start=1):
            if isinstance(arg, str):
                takes_path = ((command == "import" and i == 2)
                              or (command == "export" and i >= 2 and arg != "--with-server")
                              or (mods_add and i >= 3))
                if takes_path or "\0" in arg:
                    raise HttpError(400, f"'{command}' takes files sent along with the request, "
                                         "not paths on the daemon's host")
                argv.append(arg)
                shown.append(arg)
            elif command == "import" and i == 2 and set(arg) == {"file"}:
                item = self.server.transfers.take(arg["file"])
                cleanup.append(item["dir"])
                argv.append(item["path"])
                shown.append("<file>")
            elif mods_add and i >= 3 and set(arg) == {"dir"}:
                item = self.server.transfers.take(arg["dir"])
                cleanup.append(item["dir"])
                dest = self.server.transfers.new_dir()
                cleanup.append(dest)
                extract_upload(item["path"], dest)
                entries = sorted(os.listdir(dest))
                if not entries:
                    raise HttpError(400, "no files were sent")
                argv += [os.path.join(dest, e) for e in entries]
                shown += entries
            elif command == "export" and i >= 2 and set(arg) == {"output"} and output is None:
                if not isinstance(arg["output"], str) or not OUTPUT_RE.fullmatch(arg["output"]):
                    raise HttpError(400, "bad output file name")
                dest = self.server.transfers.new_dir()
                cleanup.append(dest)
                output = os.path.join(dest, arg["output"])
                argv.append(output)
                shown.append("<file>")
            else:
                raise HttpError(400, "bad request")
        return argv, shown, output

    def run_command(self):
        """POST /1.0/exec: runs terractl and streams its output as JSON lines:
        {"o": stdout text}, {"e": stderr text}, {"p": 1} (keepalive), and last
        {"x": exit status, "file": {...}} (an export to fetch, if any)."""
        req = self.read_json()
        via = req.get("via") or ""
        if not isinstance(via, str) or (via and not VIA_RE.fullmatch(via)):
            raise HttpError(400, "bad request")
        cleanup = []
        try:
            argv, shown, output = self.resolve(req.get("args"), cleanup)
            proc = subprocess.Popen([TERRACTL, *argv], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE, cwd="/", env=child_env(via),
                                    start_new_session=True)
            self.server.track(proc)
            dlog(f"{self.who}: terractl {shlex.join(shown)}")
            try:
                self.send_response(200)
                self.send_header("Content-Type", "application/x-ndjson")
                self.send_header("Connection", "close")
                self.end_headers()
                self.close_connection = True
                self.started = self.streaming = True
                if not self.pump(proc):
                    dlog(f"{self.who}: went away; stopped terractl {shown[0]}")
                    return
                final = {"x": proc.wait()}
                if output and final["x"] == 0 and os.path.isfile(output):
                    cleanup.remove(os.path.dirname(output))
                    final["file"] = {"id": self.server.transfers.add(output),
                                     "name": os.path.basename(output), "size": os.path.getsize(output)}
                self.frame(final)
            finally:
                self.server.untrack(proc)
        finally:
            for path in cleanup:
                shutil.rmtree(path, ignore_errors=True)

    def pump(self, proc):
        """Sends a process's output as it comes. False if the client went away."""
        streams = {proc.stdout.fileno(): ["o", codecs.getincrementaldecoder("utf-8")("replace")],
                   proc.stderr.fileno(): ["e", codecs.getincrementaldecoder("utf-8")("replace")]}
        sock = self.connection
        quiet_since = time.monotonic()
        while streams:
            ready, _, _ = select.select([*streams, sock], [], [], PING_SECONDS)
            if sock in ready and self.client_left():
                return False
            for fd in ready:
                if fd is sock:
                    continue
                key, decoder = streams[fd]
                data = os.read(fd, 65536)
                text = decoder.decode(data, final=not data)
                if text:
                    self.frame({key: text})
                    quiet_since = time.monotonic()
                if not data:
                    del streams[fd]
            if time.monotonic() - quiet_since >= PING_SECONDS:
                self.frame({"p": 1})
                quiet_since = time.monotonic()
        return True

    def client_left(self):
        sock = self.connection
        sock.settimeout(0)
        try:
            return sock.recv(4096) == b""
        except (ssl.SSLWantReadError, BlockingIOError):
            return False
        except OSError:
            return True
        finally:
            sock.settimeout(IO_TIMEOUT)

    def put_file(self):
        length = self.content_length()
        dest = self.server.transfers.new_dir()
        path = os.path.join(dest, "upload")
        try:
            self.receive(path, length)
        except BaseException:
            shutil.rmtree(dest, ignore_errors=True)
            raise
        self.send_json(200, {"id": self.server.transfers.add(path)})

    def console(self, name):
        """A line for a server's console (attach), or {"check": true} to ask
        whether the console is there."""
        if not NAME_RE.fullmatch(name):
            raise HttpError(400, f"invalid instance name '{name}'")
        req = self.read_json()
        fifo = os.path.join(self.server.run_dir, f"{name}.stdin")
        try:
            is_fifo = stat.S_ISFIFO(os.stat(fifo).st_mode)
        except FileNotFoundError:
            is_fifo = False
        if not is_fifo:
            raise HttpError(409, f"{fifo} not found - is terraria@{name} running?")
        if req.get("check"):
            return self.send_json(200, {})
        line = req.get("line")
        if not isinstance(line, str) or "\n" in line or "\r" in line or len(line) > 4096:
            raise HttpError(400, "bad console line")
        try:
            fd = os.open(fifo, os.O_WRONLY | os.O_NONBLOCK)
        except OSError as e:
            if e.errno == errno.ENXIO:
                raise HttpError(409, f"nothing reads {fifo}; is terraria@{name} running?")
            raise
        try:
            os.write(fd, (line + "\n").encode())
        finally:
            os.close(fd)
        self.send_json(200, {})

    def zip_path(self, name):
        if not ZIP_RE.fullmatch(name):
            raise HttpError(400, f"invalid zip name '{name}'")
        dist = one(self.server.info(), "dist_dir")
        return dist, os.path.join(dist, name)

    def get_zip(self, name):
        dist, path = self.zip_path(name)
        if not os.path.isfile(path):
            raise HttpError(404, f"no {name} in {dist}")
        self.send_file(path)

    def put_zip(self, name):
        dist, path = self.zip_path(name)
        if os.path.exists(path):
            raise HttpError(409, f"{name} is already in {dist}")
        length = self.content_length()
        part = os.path.join(dist, f".{name}.{secrets.token_hex(4)}.part")
        try:
            self.receive(part, length)
            os.chmod(part, 0o644)
            try:
                os.link(part, path)          # never replaces an existing zip
            except FileExistsError:
                raise HttpError(409, f"{name} is already in {dist}")
        finally:
            try:
                os.unlink(part)
            except FileNotFoundError:
                pass
        dlog(f"{self.who}: added {path}")
        self.send_json(200, {})


class Daemon(socketserver.ThreadingMixIn, http.server.HTTPServer):
    daemon_threads = True

    def __init__(self, sock, context):
        super().__init__(sock.getsockname()[:2], Handler, bind_and_activate=False)
        self.socket.close()
        self.socket = sock
        self.context = context
        self.slots = threading.BoundedSemaphore(MAX_CONNECTIONS)
        self.transfers = Transfers(state_path("transfers"))
        self.children = set()
        self.children_lock = threading.Lock()
        try:
            self.run_dir = one(self.info(), "run_dir") or "/run/terraria"
        except RuntimeError as e:
            die(str(e))

    def info(self):
        return local_host_info(child_env())

    def host_info(self):
        info = self.info()
        return {"api": API, "version": one(info, "version"), "hostname": one(info, "host"),
                "base_dir": one(info, "base_dir"), "dist_dir": one(info, "dist_dir"),
                "dist_zips": info.get("dist_zip", []), "instances": info.get("instance", [])}

    def process_request(self, request, client_address):
        if not self.slots.acquire(blocking=False):
            self.shutdown_request(request)
            return
        try:
            super().process_request(request, client_address)
        except BaseException:
            self.slots.release()
            raise

    def process_request_thread(self, request, client_address):
        try:
            super().process_request_thread(request, client_address)
        finally:
            self.slots.release()

    def finish_request(self, request, client_address):
        # TLS is set up here, in the connection's own thread, so a slow or
        # silent client can't hold up everyone else
        request.settimeout(30)
        request.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
        for option, value in (("TCP_KEEPIDLE", 60), ("TCP_KEEPINTVL", 15), ("TCP_KEEPCNT", 4)):
            if hasattr(socket, option):
                request.setsockopt(socket.IPPROTO_TCP, getattr(socket, option), value)
        try:
            conn = self.context.wrap_socket(request, server_side=True)
        except (OSError, ValueError):
            return
        try:
            self.RequestHandlerClass(conn, client_address, self)
        except (ConnectionError, TimeoutError, ssl.SSLError):
            pass       # the client hung up
        finally:
            conn.close()

    def handle_error(self, request, client_address):
        dlog(f"{client_address[0]}: {sys.exc_info()[1]!r}")

    def track(self, proc):
        with self.children_lock:
            self.children.add(proc)

    def untrack(self, proc):
        """Stops a command whose client went away, with everything it started."""
        if proc.poll() is None:
            for sig, wait in ((signal.SIGTERM, 10), (signal.SIGKILL, 5)):
                try:
                    os.killpg(proc.pid, sig)
                    proc.wait(wait)
                    break
                except ProcessLookupError:
                    break
                except subprocess.TimeoutExpired:
                    continue
        for pipe in (proc.stdout, proc.stderr):
            pipe.close()
        with self.children_lock:
            self.children.discard(proc)

    def stop_children(self):
        with self.children_lock:
            children = list(self.children)
        for proc in children:
            self.untrack(proc)


def listen_socket(host, port):
    if host in ("", "::"):
        if socket.has_dualstack_ipv6():
            try:
                return socket.create_server(("", port), family=socket.AF_INET6, dualstack_ipv6=True, backlog=64)
            except OSError as e:
                if e.errno not in (errno.EAFNOSUPPORT, errno.EADDRNOTAVAIL):
                    raise
        return socket.create_server(("", port), backlog=64)
    family, _, _, _, addr = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)[0]
    return socket.create_server(addr[:2], family=family, backlog=64)


def daemon_serve():
    open_state("daemon serve")
    ensure_cert()
    host, port = listen_setting()
    try:
        sock = listen_socket(host, port)
    except OSError as e:
        hint = ""
        if e.errno == errno.EADDRINUSE:
            hint = "; another program uses it (set TERRACTL_LISTEN in /etc/default/terractl to change the port)"
        die(f"can't listen on {describe_listen(host, port)}: {e.strerror}{hint}")
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    context.load_cert_chain(state_path(CERT), state_path(KEY))
    server = Daemon(sock, context)
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))

    def expire():
        while True:
            time.sleep(600)
            server.transfers.expire()

    threading.Thread(target=expire, daemon=True).start()
    dlog(f"terractld {VERSION} listening on {describe_listen(host, port)}; "
         f"certificate SHA-256 {cert_fingerprint()}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
        server.stop_children()
    return 0


def wait_listening(host, port, seconds):
    target = {"": "127.0.0.1", "0.0.0.0": "127.0.0.1", "::": "::1"}.get(host, host)
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        try:
            socket.create_connection((target, port), 1).close()
            return True
        except OSError:
            time.sleep(0.3)
    return False


def daemon_enable():
    need_root("daemon enable")
    open_state("daemon enable")
    ensure_cert()
    proc = systemctl("enable", "--now", DAEMON_UNIT)
    if proc.returncode != 0:
        die(f"starting terractld failed: {proc.stderr.strip()} (details: journalctl -u terractld)")
    host, port = listen_setting()
    if not wait_listening(host, port, 10):
        die(f"terractld isn't listening on {describe_listen(host, port)}; see: journalctl -u terractld")
    print(f"terractld is running and starts at boot, listening on {describe_listen(host, port)}.")
    print(f"If a firewall is active, allow TCP port {port} (e.g. sudo ufw allow {port}/tcp).")
    print()
    if load_trust()["clients"]:
        print("To add another computer: sudo terractl daemon token")
    else:
        print_token()
    return 0


def daemon_disable():
    need_root("daemon disable")
    proc = systemctl("disable", "--now", DAEMON_UNIT)
    if proc.returncode != 0:
        die(f"stopping terractld failed: {proc.stderr.strip()}")
    print("terractld is stopped and won't start at boot, so other hosts can't manage this one.")
    print("The computers it trusts are kept for next time (sudo terractl daemon status).")
    return 0


def daemon_status():
    active = systemctl("is-active", DAEMON_UNIT).stdout.strip() or "unknown"
    enabled = systemctl("is-enabled", DAEMON_UNIT).stdout.strip()
    host, port = listen_setting()
    boot = "starts at boot" if enabled == "enabled" else "doesn't start at boot"
    print(f"terractld:    {active}, {boot}")
    print(f"listening on: {describe_listen(host, port)}")
    if os.path.exists(state_path(CERT)):
        print(f"certificate:  SHA-256 {cert_fingerprint()}")
    try:
        trust = load_trust()
    except PermissionError:
        print("trusted:      (run with sudo to see which computers)")
        trust = None
    if trust is not None and not trust["clients"]:
        print("trusted:      none yet (add one: sudo terractl daemon token)")
    elif trust is not None:
        print()
        print(f"{'TRUSTED':<24} {'ADDED':<17} FROM")
        for c in trust["clients"]:
            print(f"{c['name']:<24} {c.get('added', '-'):<17} {c.get('from', '-')}")
        unused = [t for t in trust["tokens"] if t["expires"] > time.time()]
        if unused:
            print(f"(and {len(unused)} unused token{'s' if len(unused) > 1 else ''})")
    if active != "active":
        print("Start it with: sudo terractl daemon enable")
    return 0


def print_token():
    """Creates a one-time token and prints the command that uses it."""
    open_state("daemon token")
    ensure_cert()
    secret = secrets.token_hex(32)
    with StateLock():
        trust = load_trust()
        now = time.time()
        trust["tokens"] = [t for t in trust["tokens"] if t["expires"] > now]
        trust["tokens"].append({"secret_sha256": sha256(secret), "expires": int(now + TOKEN_HOURS * 3600)})
        save_trust(trust)
    host, port = listen_setting()
    fields = {"v": 1, "fp": cert_fingerprint(), "s": secret, "a": host_addresses(host, port)}
    here = (socket.gethostname() or "").split(".")[0]
    if NAME_RE.fullmatch(here):
        fields["n"] = here            # the remote's name, unless the other computer picks one
    print(f"To manage this host from another computer, run this there (works once, within {TOKEN_HOURS} hours):")
    print()
    print(f"  terractl remote add {make_token(fields)}")
    print()
    if systemctl("is-active", DAEMON_UNIT).stdout.strip() != "active":
        print("The daemon isn't running yet; start it with: sudo terractl daemon enable")
    return 0


def daemon_revoke(name):
    open_state(f"daemon revoke {name}")
    with StateLock():
        trust = load_trust()
        clients = [c for c in trust["clients"] if c["name"] != name]
        if len(clients) == len(trust["clients"]):
            die(f"no computer called '{name}' is trusted (see: sudo terractl daemon status)")
        trust["clients"] = clients
        save_trust(trust)
    print(f"'{name}' can't manage this host any more.")
    return 0


def daemon_main(argv):
    sub, rest = (argv[0], argv[1:]) if argv else ("", [])
    if sub == "revoke" and len(rest) == 1:
        return daemon_revoke(rest[0])
    commands = {"enable": daemon_enable, "disable": daemon_disable, "status": daemon_status,
                "token": print_token, "serve": daemon_serve}
    if sub in commands and not rest:
        return commands[sub]()
    die("usage: terractl daemon enable | disable | status | token | revoke <name>")


# --------------------------------------------------------------- client ----

def config_location():
    """This user's settings folder, and under sudo the account that owns it:
    remotes belong to the person, even when they use sudo."""
    if os.environ.get("TERRACTL_CONFIG_DIR"):
        return os.environ["TERRACTL_CONFIG_DIR"], None
    user = os.environ.get("SUDO_USER")
    if os.geteuid() == 0 and user and user != "root":
        try:
            pw = pwd.getpwnam(user)
            return os.path.join(pw.pw_dir, ".config", "terractl"), pw
        except KeyError:
            pass
    base = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
    return os.path.join(base, "terractl"), None


def load_remotes():
    folder, _ = config_location()
    path = os.path.join(folder, "remotes.json")
    try:
        with open(path) as f:
            return json.load(f).get("remotes", {})
    except FileNotFoundError:
        return {}
    except (OSError, ValueError, AttributeError) as e:
        die(f"can't read {path}: {e}")


def save_remotes(remotes):
    folder, owner = config_location()
    created = [d for d in (os.path.dirname(folder), folder) if not os.path.isdir(d)]
    os.makedirs(folder, mode=0o700, exist_ok=True)
    path = os.path.join(folder, "remotes.json")
    tmp = path + ".new"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump({"remotes": remotes}, f, indent=2, sort_keys=True)
        f.write("\n")
    if owner:
        for p in created + [tmp]:
            os.chown(p, owner.pw_uid, owner.pw_gid)
    os.replace(tmp, path)


class PinMismatch(Exception):
    """The daemon's certificate isn't the one pinned for it."""


class PinnedHTTPSConnection(http.client.HTTPSConnection):
    """HTTPS to a daemon whose certificate must have a known SHA-256 fingerprint."""

    def __init__(self, host, port, fingerprint, timeout):
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE        # pinned instead: see connect()
        context.minimum_version = ssl.TLSVersion.TLSv1_2
        super().__init__(host, port, timeout=timeout, context=context, blocksize=CHUNK)
        self.pin = fingerprint
        self.pin_context = context

    def connect(self):
        sock = socket.create_connection((self.host, self.port), self.timeout)
        try:
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            tls = self.pin_context.wrap_socket(sock)
        except BaseException:
            sock.close()
            raise
        got = hashlib.sha256(tls.getpeercert(binary_form=True) or b"").hexdigest()
        if not hmac.compare_digest(got, self.pin):
            tls.close()
            raise PinMismatch(got)
        self.sock = tls


class LinkLost(Exception):
    pass


class Remote:
    """A daemon on another host, as this user added it with 'terractl remote add'."""

    def __init__(self, name, conf, via=None):
        self.name = self.label = name
        self.host = conf["address"]
        self.port = int(conf["port"])
        self.fingerprint = conf["fingerprint"]
        self.key = conf.get("key", "")
        self.addr = join_hostport(self.host, self.port)
        self.via = via or f"--host {name}"
        self.console_conn = None
        self.info = None
        self.joining = False

    def same_host(self, other):
        return (self.host.lower(), self.port) == (other.host.lower(), other.port)

    def connect(self, timeout):
        conn = PinnedHTTPSConnection(self.host, self.port, self.fingerprint, timeout)
        try:
            conn.connect()
        except PinMismatch as e:
            if self.joining:
                raise Unreachable(f"the daemon at {self.addr} isn't the host the token is from (its "
                                  f"certificate's SHA-256 is {e}, the token expects {self.fingerprint})")
            raise Unreachable(
                f"the daemon at {self.addr} isn't the one '{self.name}' was added with: its certificate "
                f"changed (SHA-256 {e}, expected {self.fingerprint}). If it was replaced on purpose, add "
                f"the remote again (terractl remote remove {self.name}, then terractl remote add ...). "
                f"If not, something may be intercepting the connection.")
        except OSError as e:
            raise Unreachable(f"can't reach {self.label} at {self.addr}: {e.strerror or e}. Is its "
                              f"daemon running? (check there with: terractl daemon status)")
        return conn

    def request(self, method, path, body=None, *, timeout=30, length=None, conn=None, auth=True):
        """Sends a request and returns the response, or fails with the daemon's
        error message. Without conn, the connection is closed after the response."""
        headers = {"User-Agent": f"terractl/{VERSION}"}
        if conn is None:
            conn = self.connect(timeout)
            headers["Connection"] = "close"
            reused = False
        else:
            reused = True
        if auth:
            headers["Authorization"] = f"Bearer {self.key}"
        if isinstance(body, dict):
            body = json.dumps(body).encode()
            headers["Content-Type"] = "application/json"
        if length is not None:
            headers["Content-Length"] = str(length)
        try:
            conn.request(method, path, body=body, headers=headers)
            resp = conn.getresponse()
        except (http.client.HTTPException, OSError, PinMismatch) as e:
            conn.close()
            if reused:
                raise LinkLost()
            die(f"lost the connection to {self.label} ({e})")
        if resp.status >= 400:
            raw = resp.read()
            conn.close()
            try:
                message = json.loads(raw)["error"]
            except (ValueError, KeyError, TypeError):
                message = f"HTTP error {resp.status}"
            if resp.status == 401:
                die(f"{self.label} doesn't trust this computer (any more). Run 'sudo terractl daemon "
                    f"token' there, then here: terractl remote add <token>")
            die(f"{self.label}: {message}")
        return resp

    def call(self, method, path, body=None, **kwargs):
        resp = self.request(method, path, body, **kwargs)
        try:
            data = resp.read()
        except (http.client.HTTPException, OSError) as e:
            die(f"lost the connection to {self.label} ({e})")
        return json.loads(data or b"{}")

    def host_info(self):
        if self.info is None:
            info = self.call("GET", "/1.0")
            if info.get("api") != API:
                die(f"{self.label} runs terractl {info.get('version', '?')}, which can't work with this "
                    f"one ({VERSION}); install the same version on both")
            self.info = info
        return self.info

    def exec(self, args, show_stdout=True, capture=False):
        """Runs 'terractl <args>' on the remote, showing its output as it comes.
        Returns (exit status, last message, captured stdout)."""
        resp = self.request("POST", "/1.0/exec", {"args": args, "via": self.via}, timeout=STREAM_TIMEOUT)
        captured = []
        try:
            while True:
                try:
                    line = resp.readline()
                    frame = json.loads(line) if line else None
                except (http.client.HTTPException, OSError, ValueError) as e:
                    die(f"lost the connection to {self.label} ({e})")
                if frame is None:
                    die(f"lost the connection to {self.label} during '{args[0]}'")
                if "o" in frame:
                    if capture:
                        captured.append(frame["o"])
                    elif show_stdout:
                        sys.stdout.write(frame["o"])
                        sys.stdout.flush()
                elif "e" in frame:
                    sys.stderr.write(frame["e"])
                    sys.stderr.flush()
                elif "x" in frame:
                    return frame["x"], frame, "".join(captured)
                elif "error" in frame:
                    die(f"{self.label}: {frame['error']}")
        finally:
            resp.close()

    def upload(self, path, stream, size, label):
        """PUTs size bytes from stream, showing progress. Returns the reply."""
        progress = Progress(label, size)
        resp = self.request("PUT", path, Counted(stream, progress), length=size, timeout=IO_TIMEOUT)
        reply = json.loads(resp.read() or b"{}")
        progress.done()
        return reply

    def send_file(self, local_path, label):
        self.host_info()       # fails early, before a long upload, if we're not let in
        with open(local_path, "rb") as f:
            return self.upload("/1.0/files", f, os.fstat(f.fileno()).st_size, label)["id"]

    def console_line(self, name, line):
        """Sends a line to a server's console, over one connection kept for attach."""
        for attempt in (1, 2):
            if self.console_conn is None:
                self.console_conn = self.connect(30)
            try:
                return self.call("POST", f"/1.0/console/{name}", {"line": line}, conn=self.console_conn)
            except LinkLost:
                self.console_conn = None
                if attempt == 2:
                    die(f"lost the connection to {self.label}")


class Progress:
    """'sending x: 12.3 MB of 45.6 MB (27%)', updated in place on a terminal."""

    def __init__(self, label, total):
        self.label, self.total, self.count = label, total, 0
        self.tty = sys.stderr.isatty()
        self.shown = 0.0

    def add(self, n):
        self.count += n
        now = time.monotonic()
        if self.tty and now - self.shown >= 0.25:
            self.shown = now
            percent = self.count * 100 // self.total if self.total else 100
            sys.stderr.write(f"\r  {self.label}: {human_size(self.count)} of "
                             f"{human_size(self.total)} ({percent}%)\033[K")
            sys.stderr.flush()

    def done(self):
        line = f"  {self.label}: {human_size(self.count)}"
        sys.stderr.write(f"\r{line}\033[K\n" if self.tty else f"{line}\n")
        sys.stderr.flush()


class Counted:
    """A readable stream that reports progress as it's read."""

    def __init__(self, stream, progress):
        self.stream, self.progress = stream, progress

    def read(self, n=-1):
        data = self.stream.read(CHUNK if n is None or n < 0 else n)
        self.progress.add(len(data))
        return data


def copy_stream(src, dst, progress, size):
    while True:
        data = src.read(CHUNK)
        if not data:
            break
        dst.write(data)
        progress.add(len(data))
    if progress.count != size:
        die(f"the transfer was cut off ({human_size(progress.count)} of {human_size(size)})")
    progress.done()


def find_remote(spec):
    """A remote by name, or by the address it was added with."""
    remotes = load_remotes()
    if spec in remotes:
        return Remote(spec, remotes[spec], via=f"--host {spec}")
    try:
        host, port = split_hostport(spec)
    except Fail:
        host = port = None
    for name, conf in sorted(remotes.items()):
        if host and conf["address"].lower() == host.lower() and int(conf["port"]) == port:
            return Remote(name, conf, via=f"--host {spec}")
    die(f"'{spec}' isn't one of your remotes (see: terractl remote list). To add it, run "
        f"'sudo terractl daemon token' there, then here: terractl remote add <token>")


def remote_add(args):
    usage = "usage: terractl remote add [<name>] <token> [--address <host>[:<port>]]"
    address, rest = None, []
    args = iter(args)
    for arg in args:
        if arg == "--address":
            address = next(args, None) or die(usage)
        elif arg.startswith("--address="):
            address = arg.split("=", 1)[1]
        elif arg.startswith("-"):
            die(f"unknown option '{arg}'")
        else:
            rest.append(arg)
    if len(rest) not in (1, 2):
        die(usage)
    name, token = (None, rest[0]) if len(rest) == 1 else rest
    fields = read_token(token)
    remotes = load_remotes()
    # Adding a host that's a remote already (same certificate) replaces that
    # remote: after its access was revoked, say, or to rename it
    old = next((n for n, c in sorted(remotes.items()) if c["fingerprint"] == fields["fp"]), None)
    name = name or old or fields.get("n")
    if not name:
        die("give the remote a name: terractl remote add <name> <token>")
    if not NAME_RE.fullmatch(name):
        die(f"invalid remote name '{name}' (letters, digits and _ . -, not starting with . or -)")
    if name in remotes and name != old:
        die(f"'{name}' is already the name of another remote; choose another: terractl remote add <name> <token>")
    candidates = [split_hostport(address)] if address else [split_hostport(a) for a in fields.get("a", [])]
    if not candidates:
        die("the token doesn't say where the host is; add --address <host>")
    join = {"secret": fields["s"], "name": (socket.gethostname() or "").split(".")[0]}
    if old:
        join["old_key"] = remotes[old].get("key", "")
    problems = []
    for host, port in candidates:
        remote = Remote(name, {"address": host, "port": port, "fingerprint": fields["fp"]})
        remote.joining = True
        remote.label = remote.addr
        try:
            reply = remote.call("POST", "/1.0/trust/join", join, auth=False, timeout=5)
            break
        except Unreachable as e:
            problems.append(str(e))
    else:
        if len(problems) == 1:
            die(problems[0])
        die("couldn't reach the host at any address in the token:\n  " + "\n  ".join(problems) +
            "\nGive an address that works from here: terractl remote add <token> --address <host>")
    if old:
        del remotes[old]
    remotes[name] = {"address": host, "port": port, "fingerprint": fields["fp"], "key": reply["key"],
                     "client_name": reply["name"], "added": time.strftime("%Y-%m-%d %H:%M")}
    save_remotes(remotes)
    remote.key = reply["key"]
    remote.joining = False
    remote.label = name
    try:
        info = remote.host_info()
        about = f"{info['hostname']} at {remote.addr}, terractl {info['version']}"
    except Fail:
        about = remote.addr
    print(f"{'updated' if old else 'added'} remote '{name}' ({about}); it knows this computer as '{reply['name']}'")
    print(f"Try: terractl --host {name} list")
    return 0


def remote_list(args):
    if args:
        die("usage: terractl remote list")
    remotes = load_remotes()
    if not remotes:
        print("No remotes yet. Run 'sudo terractl daemon token' on a host, then here: terractl remote add <token>")
        return 0
    print(f"{'NAME':<20} {'ADDRESS':<30} FINGERPRINT")
    for name, conf in sorted(remotes.items()):
        print(f"{name:<20} {join_hostport(conf['address'], conf['port']):<30} {conf['fingerprint'][:16]}")
    return 0


def remote_remove(args):
    if len(args) != 1:
        die("usage: terractl remote remove <name>")
    remotes = load_remotes()
    conf = remotes.pop(args[0], None)
    if conf is None:
        die(f"there's no remote called '{args[0]}' (see: terractl remote list)")
    save_remotes(remotes)
    print(f"removed remote '{args[0]}'. It still trusts this computer; to revoke that too, run there: "
          f"sudo terractl daemon revoke {conf.get('client_name', '<name>')}")
    return 0


def remote_main(argv):
    sub, rest = (argv[0], argv[1:]) if argv else ("list", [])
    commands = {"add": remote_add, "list": remote_list, "remove": remote_remove}
    if sub not in commands:
        die("usage: terractl remote add <name> <token> | remote list | remote remove <name>")
    return commands[sub](rest)


# ------------------------------------------------ commands on a remote ----

def local_path(path):
    return os.path.abspath(os.path.expanduser(path))


def on_remote_attach(remote, argv):
    if len(argv) != 2:
        die("usage: terractl attach <name>")
    name = argv[1]
    if not NAME_RE.fullmatch(name):
        die(f"invalid instance name '{name}'")
    remote.call("POST", f"/1.0/console/{name}", {"check": True})
    print(f"Attached to terraria@{name} on {remote.label}. Ctrl-D or Ctrl-C detaches; the server keeps running.")

    def follow():
        try:
            remote.exec(["watch", name, "-n", "30", "-q"])
        except Fail as e:
            log(str(e))
        log(f"no more output from {remote.label}; Ctrl-D detaches")

    threading.Thread(target=follow, daemon=True).start()
    try:
        import readline          # line editing and history for input()
        readline.set_auto_history(True)
    except ImportError:
        pass
    try:
        while True:
            line = input()
            if not line:
                continue
            if line == "exit":
                print(f"(not sent: 'exit' would shut the server down. Use Ctrl-D to detach, "
                      f"or 'terractl {remote.via} stop {name}' to stop it.)")
                continue
            try:
                remote.console_line(name, line)
            except Fail as e:
                print(f"(not sent: {e})", file=sys.stderr)
    except (EOFError, KeyboardInterrupt):
        print()
    return 0


def on_remote_watch(remote, argv):
    try:
        return remote.exec(argv)[0]
    except KeyboardInterrupt:
        print()
        return 0


def on_remote_export(remote, argv):
    if len(argv) < 2:
        die("usage: terractl export <name> [file.tar.gz] [--with-server]")
    name, out, with_server = argv[1], None, False
    if not NAME_RE.fullmatch(name):
        die(f"invalid instance name '{name}'")
    for arg in argv[2:]:
        if arg == "--with-server":
            with_server = True
        elif arg.startswith("-"):
            die(f"unknown option '{arg}'")
        elif out:
            die("more than one output file given")
        else:
            out = arg
    out = local_path(out or f"{name}-{time.strftime('%Y%m%d-%H%M%S')}.tar.gz")
    if not out.endswith((".tar.gz", ".tgz")):
        out += ".tar.gz"
    if os.path.exists(out):
        die(f"{out} already exists")
    if not os.path.isdir(os.path.dirname(out)):
        die(f"folder {os.path.dirname(out)} does not exist")
    archive = RemoteHost(remote).export(name, with_server)
    archive.save(out, f"fetching {archive.name}")
    print(f"exported '{name}' from {remote.label} to {out} ({human_size(archive.size)}, readable only by you)")
    if not with_server:
        print("server/ not included; it's unpacked again from the server zip on first start after import")
    return 0


def on_remote_import(remote, argv):
    if len(argv) != 3:
        die("usage: terractl import <new instance name> <file.tar.gz>")
    path = local_path(argv[2])
    if not os.path.isfile(path):
        die(f"cannot read {path}")
    fid = remote.send_file(path, f"sending {os.path.basename(path)} to {remote.label}")
    return remote.exec(["import", argv[1], {"file": fid}])[0]


def on_remote_mods(remote, argv):
    if len(argv) < 4 or argv[2] != "add":
        return remote.exec(argv)[0]
    tmp = big_tempdir()
    try:
        bundle = os.path.join(tmp, "mods.tar")
        names = set()
        with tarfile.open(bundle, "w", dereference=True) as tar:
            for arg in argv[3:]:
                path = local_path(arg)
                if not os.path.exists(path):
                    die(f"{path} not found")
                base = os.path.basename(os.path.normpath(path))
                if base in names:
                    die(f"more than one of the given paths is called '{base}'")
                names.add(base)
                tar.add(path, arcname=base, filter=lambda m: m if m.isfile() or m.isdir() else None)
        fid = remote.send_file(bundle, f"sending mods to {remote.label}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)
    return remote.exec(["mods", argv[1], "add", {"dir": fid}])[0]


def on_remote_delete(remote, argv):
    if len(argv) >= 2 and "-y" not in argv and "--yes" not in argv:
        name = argv[1]
        if not sys.stdin.isatty():
            die("not a terminal; use -y to delete without confirmation")
        print(f"This will stop terraria@{name} on {remote.label} and remove it there.")
        if "--no-backup" in argv:
            print("NO BACKUP will be made. Worlds will be lost.")
        else:
            print(f"Worlds and configs will be archived to {remote.host_info()['base_dir']}/.deleted/ "
                  f"on {remote.label} first.")
        if input("Type the instance name to confirm: ") != name:
            die("confirmation did not match; nothing deleted")
        argv = argv + ["-y"]
    return remote.exec(argv)[0]


def run_on_remote(spec, argv):
    if not argv:
        die("usage: terractl --host <remote> <command> [args]")
    remote = find_remote(spec)
    command = argv[0]
    if command in ("remote", "daemon", "copy", "move", "-H") or command.startswith("--host"):
        die(f"'{command}' runs on this computer; use it without --host "
            "(copy and move take <remote>:<name> instead)")
    if command in ("prepare", "run", "service-stop", "journal-access"):
        die(f"'{command}' is only used by the service itself")
    handlers = {"attach": on_remote_attach, "watch": on_remote_watch, "export": on_remote_export,
                "import": on_remote_import, "mods": on_remote_mods, "delete": on_remote_delete}
    if command in handlers:
        return handlers[command](remote, argv)
    return remote.exec(argv)[0]


# ------------------------------------------------------------ copy/move ----

class LocalArchive:
    def __init__(self, path, tmpdir):
        self.path, self.tmpdir = path, tmpdir
        self.name = os.path.basename(path)

    def cleanup(self):
        shutil.rmtree(self.tmpdir, ignore_errors=True)


class RemoteArchive:
    """An export waiting on a daemon. Fetching it (or importing it on the same
    host) uses it up; otherwise cleanup() removes it there."""

    def __init__(self, host, fid, name, size):
        self.host, self.fid, self.name, self.size = host, fid, name, size
        self.used = False

    def claim(self):
        self.used = True
        return self.fid

    def open(self):
        self.used = True
        return self.host.remote.request("GET", f"/1.0/files/{self.fid}", timeout=IO_TIMEOUT)

    def save(self, path, label):
        stream = self.open()
        try:
            fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            try:
                with os.fdopen(fd, "wb") as out:
                    copy_stream(stream, out, Progress(label, self.size), self.size)
            except BaseException:
                os.unlink(path)
                raise
        finally:
            stream.close()

    def cleanup(self):
        if not self.used:
            try:
                self.host.remote.call("DELETE", f"/1.0/files/{self.fid}")
            except Fail:
                pass


class LocalHost:
    label = "this host"
    is_local = True

    def same_host(self, other):
        return isinstance(other, LocalHost)

    def host_info(self):
        info = parse_info(terractl(["info"], capture=True).stdout)
        return {"hostname": one(info, "host"), "base_dir": one(info, "base_dir"),
                "dist_dir": one(info, "dist_dir"), "dist_zips": info.get("dist_zip", [])}

    def instance(self, name):
        base = self.host_info()["base_dir"]
        if not os.access(base, os.R_OK | os.X_OK):
            die(f"can't read the instances in {base}: join the terraria group, or use sudo")
        proc = terractl(["info", name], capture=True)
        if proc.returncode != 0:
            die(f"can't read instance '{name}' on this host")
        return {k: v[-1] for k, v in parse_info(proc.stdout).items()}

    def run(self, args, show_stdout=True):
        return terractl(args, show_stdout=show_stdout).returncode

    def export(self, name, with_server):
        tmp = big_tempdir()
        path = os.path.join(tmp, f"{name}.tar.gz")
        if self.run(["export", name, path] + (["--with-server"] if with_server else []), show_stdout=False):
            shutil.rmtree(tmp, ignore_errors=True)
            die(f"exporting '{name}' failed")
        return LocalArchive(path, tmp)

    def import_(self, name, archive, show_stdout):
        if isinstance(archive, LocalArchive):
            return self.run(["import", name, archive.path], show_stdout)
        tmp = big_tempdir()
        try:
            path = os.path.join(tmp, archive.name)
            archive.save(path, f"fetching {archive.name} from {archive.host.label}")
            return self.run(["import", name, path], show_stdout)
        finally:
            shutil.rmtree(tmp, ignore_errors=True)

    def open_zip(self, zip_name):
        f = open(os.path.join(self.host_info()["dist_dir"], zip_name), "rb")
        return f, os.fstat(f.fileno()).st_size

    def receive_zip(self, zip_name, stream, size, label):
        dist = self.host_info()["dist_dir"]
        part = os.path.join(dist, f".{zip_name}.{secrets.token_hex(4)}.part")
        try:
            with open(part, "xb") as out:
                copy_stream(stream, out, Progress(label, size), size)
            os.chmod(part, 0o644)
            os.link(part, os.path.join(dist, zip_name))      # never replaces an existing zip
        finally:
            try:
                os.unlink(part)
            except FileNotFoundError:
                pass


class RemoteHost:
    is_local = False

    def __init__(self, remote):
        self.remote = remote
        self.label = remote.label

    def same_host(self, other):
        return isinstance(other, RemoteHost) and self.remote.same_host(other.remote)

    def host_info(self):
        return self.remote.host_info()

    def instance(self, name):
        code, _, out = self.remote.exec(["info", name], capture=True)
        if code != 0:
            die(f"can't read instance '{name}' on {self.label}")
        return {k: v[-1] for k, v in parse_info(out).items()}

    def run(self, args, show_stdout=True):
        return self.remote.exec(args, show_stdout=show_stdout)[0]

    def export(self, name, with_server):
        args = ["export", name, {"output": f"{name}.tar.gz"}] + (["--with-server"] if with_server else [])
        code, final, _ = self.remote.exec(args, show_stdout=False)
        if code != 0 or "file" not in final:
            die(f"exporting '{name}' on {self.label} failed")
        f = final["file"]
        return RemoteArchive(self, f["id"], f["name"], f["size"])

    def import_(self, name, archive, show_stdout):
        if isinstance(archive, RemoteArchive) and archive.host.same_host(self):
            fid = archive.claim()                    # already there
        elif isinstance(archive, LocalArchive):
            fid = self.remote.send_file(archive.path, f"sending {archive.name} to {self.label}")
        else:
            self.host_info()
            stream = archive.open()
            try:
                fid = self.remote.upload("/1.0/files", stream, archive.size,
                                         f"copying {archive.name} from {archive.host.label} to {self.label}")["id"]
            finally:
                stream.close()
        return self.remote.exec(["import", name, {"file": fid}], show_stdout=show_stdout)[0]

    def open_zip(self, zip_name):
        resp = self.remote.request("GET", f"/1.0/dist/{urllib.parse.quote(zip_name)}", timeout=IO_TIMEOUT)
        return resp, int(resp.getheader("Content-Length") or 0)

    def receive_zip(self, zip_name, stream, size, label):
        self.remote.upload(f"/1.0/dist/{urllib.parse.quote(zip_name)}", stream, size, label)


def transfer_label(src, dst, name):
    if dst.is_local:
        return f"fetching {name} from {src.label}"
    if src.is_local:
        return f"sending {name} to {dst.label}"
    return f"copying {name} from {src.label} to {dst.label}"


def parse_location(text):
    """'[<remote>:]<name>' -> (host, name). The name may be empty ('nas:')."""
    if ":" not in text:
        return LocalHost(), text
    spec, _, name = text.rpartition(":")
    return RemoteHost(find_remote(spec)), name


def copy_or_move(argv, move):
    verb = "move" if move else "copy"
    usage = f"usage: terractl {verb} [<remote>:]<name> [<remote>:][<new name>] [--with-server]"
    with_server, places = False, []
    for arg in argv:
        if arg == "--with-server":
            with_server = True
        elif arg.startswith("-"):
            die(f"unknown option '{arg}'")
        else:
            places.append(arg)
    if len(places) != 2:
        die(usage)
    src, name = parse_location(places[0])
    dst, new_name = parse_location(places[1])
    new_name = new_name or name
    for n in (name, new_name):
        if not NAME_RE.fullmatch(n):
            die(f"invalid instance name '{n}'" if n else usage)
    if src.same_host(dst) and name == new_name:
        die("the source and the destination are the same")
    if move and src.same_host(dst):
        # Renaming in place keeps the port, server/ and tModLoader's .NET
        if src.is_local and os.geteuid() != 0:
            die(f"renaming needs root: sudo terractl move {shlex.join(argv)}")
        return src.run(["rename", name, new_name])

    # Check everything first, so nothing changes if it can't work
    if move and src.is_local and os.geteuid() != 0:
        die(f"moving '{name}' deletes it on this host, which needs root: sudo terractl move {shlex.join(argv)}")
    if dst.is_local and os.geteuid() != 0 and not os.access(dst.host_info()["base_dir"], os.W_OK):
        die(f"adding an instance on this host needs root: sudo terractl {verb} {shlex.join(argv)}")
    source = src.instance(name)
    if source.get("exists") != "yes":
        die(f"no instance '{name}' on {src.label}")
    if dst.instance(new_name).get("exists") == "yes":
        prefix = places[1].rpartition(":")[0] + ":" if ":" in places[1] else ""
        die(f"'{new_name}' already exists on {dst.label}; give it another name there, "
            f"e.g. terractl {verb} {places[0]} {prefix}<new name>")
    # The server zip the instance runs (import pins it). It has to be in the
    # destination's dist folder, unless the unpacked server goes along.
    needed = os.path.basename(source.get("pinned_zip", "")) or source.get("installed_zip", "")
    send_zip = None
    if needed and not with_server and not src.same_host(dst):
        dst_info = dst.host_info()
        if needed not in dst_info["dist_zips"]:
            if needed not in src.host_info()["dist_zips"]:
                die(f"'{name}' runs {needed}, which {dst.label} doesn't have in {dst_info['dist_dir']} and "
                    f"{src.label} has no copy of to send. Put it there first, or add --with-server to "
                    f"include the unpacked server")
            send_zip = needed

    was_running = source.get("state") in ("active", "activating", "reloading")
    was_enabled = source.get("boot") == "enabled"
    to = f" as '{new_name}'" if new_name != name else ""
    print(f"{'Moving' if move else 'Copying'} '{name}' from {src.label} to {dst.label}{to}")
    stopped = imported = False
    archive = None
    try:
        if move and was_running:
            print(f"stopping '{name}' on {src.label} (saves the world first)...")
            if src.run(["stop", name], show_stdout=False) != 0:
                die(f"couldn't stop '{name}' on {src.label}")
            stopped = True
        archive = src.export(name, with_server)
        if send_zip:
            stream, size = src.open_zip(send_zip)
            try:
                dst.receive_zip(send_zip, stream, size, transfer_label(src, dst, send_zip))
            finally:
                stream.close()
        if dst.import_(new_name, archive, show_stdout=not move) != 0:
            die(f"importing on {dst.label} failed; '{name}' on {src.label} is unchanged")
        imported = True
    except BaseException:
        if stopped and not imported:
            log(f"starting '{name}' on {src.label} again")
            try:
                src.run(["start", name], show_stdout=False)
            except Fail as e:
                log(f"that failed too: {e}")
        raise
    finally:
        if archive:
            archive.cleanup()
    if not move:
        return 0

    if src.run(["delete", name, "-y"]) != 0:
        die(f"'{new_name}' is on {dst.label} now, but deleting '{name}' on {src.label} failed (see above), "
            f"so it's in both places. The copy on {dst.label} wasn't started.")
    now = []
    if was_enabled and dst.run(["enable", new_name], show_stdout=False) == 0:
        now.append("starts at boot")
    if was_running and dst.run(["start", new_name], show_stdout=False) == 0:
        now.append("is running")
    state = f"; it {' and '.join(now)} there" if now else ""
    print(f"moved '{name}' from {src.label} to {dst.label}{to}{state}")
    return 0


# ---------------------------------------------------------------- main ----

def main(argv):
    if not argv:
        die("usage: terractl --host <remote> <command> | remote ... | daemon ... | copy ... | move ...")
    first = argv[0]
    if first in ("--host", "-H"):
        if len(argv) < 2:
            die("usage: terractl --host <remote> <command> [args]")
        return run_on_remote(argv[1], argv[2:])
    if first.startswith("--host="):
        return run_on_remote(first[len("--host="):], argv[1:])
    if first == "remote":
        return remote_main(argv[1:])
    if first == "daemon":
        return daemon_main(argv[1:])
    if first in ("copy", "move"):
        return copy_or_move(argv[1:], first == "move")
    die(f"unknown command '{first}'")


if __name__ == "__main__":
    sys.stdout.reconfigure(line_buffering=True)    # keeps it in order with terractl's own output
    # terractl works from / and says where it was called from (see abspath there)
    try:
        os.chdir(os.environ.get("TERRACTL_CALLER_PWD") or ".")
    except OSError:
        pass
    try:
        sys.exit(main(sys.argv[1:]))
    except Fail as e:
        print(f"terractl: {e}", file=sys.stderr)
        sys.exit(1)
    except BrokenPipeError:
        # Output went into something that stopped reading (e.g. head)
        os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
        sys.exit(1)
    except OSError as e:
        where = f": {e.filename}" if e.filename else ""
        print(f"terractl: {e.strerror or e}{where}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        print(file=sys.stderr)
        sys.exit(130)