adamandClaude Opus 5.5
fc2c08d47b
release / deb (push) Successful in 17s
- doctor [<name>] [--fix]: checks the system (user, programs, folders, zips, units and leftover copies in /etc, sandbox access to moved folders, journal, polkit, daemon, orphaned boot links and schedules) and each instance (ownership, server zip, config paths, mods, ports, failed starts/backups), with a fix for each. --fix applies the safe ones and re-checks them. - backup enable|disable|now|list: a systemd timer per instance (terraria-backup@.timer, Persistent=), saving the world first, into /var/lib/terraria/backups/<name>/ (group-readable), pruning only scheduled backups. rename carries the schedule, delete removes it. upgrade's backups move there too. - fetch vanilla|tmodloader [<version>] [--list] [--upgrade <name>]: downloads from terraria.org / tModLoader's GitHub releases, checks GitHub's sha256 and the zip, names it with its version. - terractld may write /etc/systemd/system, for backup enable and doctor --fix over --host. - Units, tmpfiles, packaging, completion, docs and changelog. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1955 lines
75 KiB
Python
Executable File
1955 lines
75 KiB
Python
Executable File
#!/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:
|
|
die("openssl is not installed (sudo apt 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)
|