public/supaspin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

feat: tunnel guide in list, push-env, VERSION file

Browse Source 
- list and tunnel print plain-language tunnel instructions for every instance: keep it open, run it in the
  background, close it, with a notice that background/close work on Mac and Linux only, not Windows
- push-env N TARGET puts an instance's .env values into another .env file, on this server or over SSH (Mac, Linux,
  Windows): existing keys are replaced, missing keys added, commented-out lines ignored, every other line kept.
  A temporary copy of the target is compared after writing and put back on any mismatch. --dry-run shows the
  changes without writing. SSH failures are explained with how to set SSH up.
- VERSION file (1.1.0), version line in the README, `--version` prints "supaspin 1.1.0"; a test keeps the file, the
  code constant and the README in step
- remove an unused parameter of run() and an unreachable second help branch in main()

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Igor K.
2026-09-19 12:22:13 +03:00
co-authored by Claude Sonnet 5
parent cfbc4f1612
commit 2af1a83cda
4 changed files with 1345 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+86 -9
View File
@@ -1,5 +1,7 @@
# supaspin
Version 1.1.0
Create, list, show and delete throwaway [Supabase](https://supabase.com) instances on a
single-node [microk8s](https://microk8s.io) server, from a terminal menu.
@@ -87,9 +89,10 @@ Supabase Helm chart (0.8.0, Postgres 17.6.1.136) once into `/var/lib/supaspin/ch
| Menu item | What it does |
|---|---|
| Create instance | picks the lowest free number, asks whether to use the saved mail settings or enter different ones, creates the disks, installs Supabase, waits until every pod is ready, writes `<root>/<N>/.env` |
| List instances | number, hostname, pods ready, memory in use, disk use per image, `.env` path |
| List instances | number, hostname, pods ready, memory in use, disk use per image, `.env` path, and for each instance the tunnel command with plain instructions (open it, keep it in the background, close it) |
| Show instance | the `.env` (secrets hidden unless you ask), the tunnel command, pod status and recent events |
| Delete instance | asks you to type the number; removes everything (see below) |
| Copy instance values into another .env file | puts an instance's `.env` values into your app's `.env`, on this server or on another computer (see below) |
| Mail settings | view and change the saved default mail settings |
| Settings | change any first-run setting |
@@ -100,10 +103,12 @@ sudo supaspin create [--ask-mail] # --ask-mail: choose saved or different ma
sudo supaspin list
sudo supaspin show N # .env with secrets hidden, plus tunnel help
sudo supaspin env N # the full .env (contains secrets)
sudo supaspin tunnel N
sudo supaspin tunnel N # the tunnel commands, step by step
sudo supaspin push-env N TARGET [--port P] [--key FILE] [--dry-run] # see "Copying an instance's values"
sudo supaspin delete N [--yes]
sudo supaspin mail # change the saved default mail settings
sudo supaspin setup # change settings
supaspin --version # prints "supaspin 1.1.0"
```
### Connecting your app
@@ -115,15 +120,80 @@ names: `SUPABASE_URL`, `PUBLIC_SUPABASE_URL`, `API_EXTERNAL_URL`, `ANON_KEY`,
`GOTRUE_SMTP_*`. Read it with `sudo supaspin env N` and copy what your app needs.
The web API is public, so the URL and keys work from anywhere. **Postgres and Studio are not
public**; from outside the server's network use an SSH tunnel:
public.** To reach them from your own computer you open a *tunnel*: a private connection from your
computer to the server that makes the instance's database and Studio (its admin screen) appear on
your own computer, at `localhost:544N` (database) and `http://localhost:300N` (Studio), for as long
as the tunnel is open. `sudo supaspin list` prints the exact commands for every instance
(`sudo supaspin tunnel N` prints them with a few more hints). Run them in a terminal on **your**
computer, not on the server.
1. On your computer, run the command from `supaspin tunnel N` in its own terminal and leave it open:
`ssh -N -p 2222 -L 5441:192.168.10.30:5441 -L 3001:192.168.10.30:3001 you@your.server`
2. Put `DATABASE_URL=postgresql://postgres:<password>@localhost:5441/postgres` in your app's
`.env` (the password is `POSTGRES_PASSWORD` in the instance's `.env`).
3. Studio is at `http://localhost:3001`.
**Which computer you use matters.** The commands for keeping the tunnel in the background and for
closing it work on **Mac and Linux only**. They do **not** work on **Windows**, because the ssh
program that comes with Windows does not support the "control file" they need. On Windows use the
first command (keep the tunnel open in its own PowerShell window) and close it with Ctrl+C or by
closing the window.
On the server's own LAN skip the tunnel and use `<LAN IP>:5441` and `http://<LAN IP>:3001`.
- If you want to open the tunnel and keep it open in your terminal window (Mac, Linux, Windows),
run the `ssh -N ...` command. The window looks stuck and prints nothing: that is normal, it means
the tunnel is open. Press Ctrl+C to close it.
- If you want the tunnel to run in the background so your terminal stays free (Mac and Linux only),
run the `ssh -f -N -M -S /tmp/supaspin-N.sock ...` command. It asks for your password if needed,
then gives you your terminal back.
- If you no longer need the background tunnel and want to close it (Mac and Linux only), run
`ssh -S /tmp/supaspin-N.sock -O exit you@your.server`. It answers "Exit request sent."
- If you want to check whether the background tunnel is running (Mac and Linux only), run
`ssh -S /tmp/supaspin-N.sock -O check you@your.server`. It answers "Master running" if it is.
- If closing says "No such file or directory" but the tunnel still works, or starting a tunnel says
"address already in use" and the check does not say "Master running", a leftover tunnel is in the
way: run `pkill -f "[s]upaspin-N.sock"` (Mac and Linux only; the `[s]` is on purpose, it stops the command from
stopping itself) and start the tunnel again.
While the tunnel is open, put `DATABASE_URL=postgresql://postgres:<password>@localhost:5441/postgres`
in your app's `.env` (the password is `POSTGRES_PASSWORD` in the instance's `.env`) and open Studio
at `http://localhost:3001`. On the server's own LAN you do not need a tunnel: use
`<LAN IP>:5441` and `http://<LAN IP>:3001`.
### Copying an instance's values into your app's .env
If you want your app's `.env` to point at an instance, run:
```bash
sudo supaspin push-env 1 igor@192.168.10.50:/Users/igor/app/.env # a Mac or Linux computer
sudo supaspin push-env 1 "igor@192.168.10.50:C:/Users/igor/app/.env" # a Windows computer
sudo supaspin push-env 1 /home/igor/app/.env # a file on this server
sudo supaspin push-env 1 --dry-run igor@192.168.10.50:/Users/igor/app/.env # only show what would change
```
or choose "Copy instance values into another .env file" in the menu, which asks for everything
(instance, file, SSH user, port and key file). Options: `--port` (SSH port, default 22), `--key`
(key file), `--dry-run` (write nothing). If the SSH server needs a password, SSH asks for it in
your terminal, once.
What it does with the target file:
- Every `KEY=VALUE` line of the instance's `.env` is put in. If the key is already in the file
(also as `export KEY=...`), its value is replaced. If it is not, the line is added at the end.
- Lines that start with `#` are ignored completely. `# ANON_KEY=old` is left as it is, and
`ANON_KEY` counts as missing, so a new `ANON_KEY=...` line is added.
- Every other line (other keys, comments, blank lines, their order, line endings, the file's owner
and permissions) stays exactly as it was. The file must already exist; it is never created.
- It never prints values, only key names: replaced, added, already correct, and how many other
lines stayed untouched. It ends with SUCCESS or FAILED and the reason, and its exit status is
0 or 1.
Safety net: before the file is changed, an untouched copy is kept in a private temporary folder
(in memory, in `/run`). After writing, the file is read back and compared with that copy: nothing
but the instance's keys may differ, and each of them must have the instance's value. If anything is
off, the copy is written back and checked. The copy is deleted at the end. Only if even the
write-back fails is it kept, and its path is printed so that you can restore the file by hand.
The other computer can be a Mac, a Linux computer or a Windows computer. It only needs an SSH server
with file copying (SFTP) switched on and this server must be able to reach it. Files are copied with
`scp`, so nothing has to be installed there. The SSH connection is made as the person who typed
`sudo`, so your own keys and known computers apply. If the connection fails, supaspin says why and
what to do: for example how to switch SSH on (Mac: System Settings > General > Sharing > Remote
Login; Linux: `sudo apt install openssh-server`; Windows: the optional feature "OpenSSH Server"),
how to install this server's key (`ssh-copy-id`), or how to trust a new computer.
### Deleting
@@ -187,6 +257,13 @@ The tests use Python's built-in `unittest` and need no extra packages:
python3 -m unittest discover -s tests
```
### Versions
The version is written in three places that must always show the same number: the file `VERSION`, the
`VERSION` constant at the top of `supaspin`, and the line "Version x.y.z" at the top of this README. A test
fails if they differ. Every release is a git tag `vX.Y.Z` (for example `v1.1.0`) on the commit that has that
number in `VERSION`.
## License
MIT, see `LICENSE`.
VERSION
+1
View File
@@ -0,0 +1 @@
1.1.0
supaspin
+655 -14
View File
@@ -5,27 +5,34 @@ Each numbered instance gets its own namespace, capped disk images, capped memory
an HTTPS hostname and fixed Postgres/Studio ports. Standard library plus python3-yaml.
Run as root: supaspin (menu)
supaspin create|list|show N|env N|tunnel N|delete N|setup|mail
supaspin push-env N TARGET [--port P] [--key FILE] [--dry-run]
TARGET = a .env file on this server (/path/.env) or on another computer
(user@host:/path/.env, Windows: user@host:C:/path/.env). Puts the instance's
KEY=VALUE pairs into it: existing keys get the new value, missing keys are added.
"""
import argparse
import base64
import glob
import hashlib
import hmac
import json
import os
import pwd
import re
import secrets
import shutil
import string
import subprocess
import sys
import tempfile
import termios
import time
import tty
import yaml
VERSION = "1.0.0"
VERSION = "1.1.0"
CONFIG_PATH = os.environ.get("SUPASPIN_CONFIG", "/etc/supaspin/config.yaml")
STATE_DIR = os.environ.get("SUPASPIN_STATE", "/var/lib/supaspin")
CHART_REPO = "https://supabase-community.github.io/supabase-kubernetes"
@@ -240,15 +247,78 @@ def tunnel_command(cfg, n):
f"-L {studio_port(n)}:{lan}:{studio_port(n)} {cfg['ssh_user']}@{cfg['ssh_host']}")
def tunnel_help(cfg, n, pw="<password>"):
return "\n".join([
"Run this in its own terminal on your computer and leave it open:",
def tunnel_socket(n):
"""The 'control file' of a background tunnel: ssh is later told to quit through it."""
return f"/tmp/supaspin-{n}.sock"
def tunnel_background_command(cfg, n):
lan = cfg["lan_ip"]
return (f"ssh -f -N -M -S {tunnel_socket(n)} -o ExitOnForwardFailure=yes -p {cfg['ssh_port']} "
f"-L {pg_port(n)}:{lan}:{pg_port(n)} -L {studio_port(n)}:{lan}:{studio_port(n)} "
f"{cfg['ssh_user']}@{cfg['ssh_host']}")
def tunnel_control_command(cfg, n, action):
"""`check` asks the background tunnel whether it runs, `exit` closes it."""
return f"ssh -S {tunnel_socket(n)} -O {action} {cfg['ssh_user']}@{cfg['ssh_host']}"
def tunnel_guide(cfg, n, detailed=False):
"""Plain-language tunnel instructions for instance n, as a list of lines."""
if not cfg.get("ssh_host"):
return ["Tunnel not available: run 'sudo supaspin setup' and fill in 'Public SSH host for tunnels'."]
lines = [
"Tunnel = a private connection from YOUR computer to this instance, so you can reach its",
f"database (localhost:{pg_port(n)}) and Studio admin screen (http://localhost:{studio_port(n)}) from your computer.",
"Run the commands below in a terminal on YOUR computer, not on this server.",
"",
"IMPORTANT - which computer you use matters:",
" Mac or Linux: all the commands below work.",
" Windows: ONLY the first command below (keep the tunnel open in the window) works. The background",
" and closing commands do NOT work on Windows, because the ssh program that comes with Windows",
' does not support the "control file" they need. On Windows keep the tunnel open in its own',
" PowerShell window and close it with Ctrl+C, or by closing that window.",
"",
"If you want to open the tunnel and keep it open in the terminal window you are using, run:",
f" {tunnel_command(cfg, n)}",
"Then, in the app's .env:",
"The window then looks stuck and prints nothing - that is normal, it means the tunnel is open.",
"To close it, press Ctrl+C in that window.",
"",
"If you want the tunnel to keep running in the background, so your terminal stays free (Mac or Linux only), run:",
f" {tunnel_background_command(cfg, n)}",
"If it asks for your password, type it. Then you get your terminal back and the tunnel stays open.",
"",
"If you no longer need the background tunnel and want to close it (Mac or Linux only), run:",
f" {tunnel_control_command(cfg, n, 'exit')}",
'It answers "Exit request sent." and the tunnel is closed.',
]
if not detailed:
return lines + ["", f"More help (is the tunnel running? what if closing fails?): sudo supaspin tunnel {n}"]
return lines + [
"",
"If you want to check whether the background tunnel is running (Mac or Linux only), run:",
f" {tunnel_control_command(cfg, n, 'check')}",
'It answers "Master running" when the tunnel is open.',
"",
'If closing says "No such file or directory" but the tunnel still works, or starting a tunnel says',
'"address already in use" and the check above does not say "Master running", a leftover tunnel is',
"in the way. To stop it (Mac or Linux only), run:",
f' pkill -f "[s]upaspin-{n}.sock"',
"(The [s] is on purpose: it stops the command from also matching, and stopping, itself.)",
"Then start the tunnel again.",
]
def tunnel_help(cfg, n, pw="<password>"):
return "\n".join(tunnel_guide(cfg, n, detailed=True) + [
"",
"When the tunnel is open, put this in your app's .env:",
f" DATABASE_URL=postgresql://postgres:{pw}@localhost:{pg_port(n)}/postgres",
" (the password is POSTGRES_PASSWORD in the instance's .env: supaspin env " + str(n) + ")",
" (the password is POSTGRES_PASSWORD in the instance's .env: sudo supaspin env " + str(n) + ")",
f"Studio (admin UI): http://localhost:{studio_port(n)}",
f"On the LAN skip the tunnel: {cfg['lan_ip']}:{pg_port(n)} and http://{cfg['lan_ip']}:{studio_port(n)}",
f"On the server's own network you do not need a tunnel: use {cfg['lan_ip']}:{pg_port(n)} "
f"and http://{cfg['lan_ip']}:{studio_port(n)}",
])
@@ -357,7 +427,7 @@ def pv_manifest(cfg, n, vol):
# ------------------------------------------------------------------ system --
def run(args, input_=None, check=True, quiet=False):
def run(args, input_=None, check=True):
p = subprocess.run(args, input=input_, capture_output=True, text=True)
if check and p.returncode != 0:
raise RuntimeError(f"{' '.join(args)}\n{(p.stderr or p.stdout).strip()}")
@@ -759,6 +829,567 @@ def list_rows(cfg):
return rows
# ------------------------------------------------------ copy .env values --
_ASSIGN = re.compile(r"^([ \t]*(?:export[ \t]+)?)([A-Za-z_][A-Za-z0-9_.\-]*)([ \t]*=[ \t]*)(.*)$")
_PAIR = re.compile(r"^[ \t]*(?:export[ \t]+)?([A-Za-z_][A-Za-z0-9_.\-]*)[ \t]*=(.*)$")
_REMOTE = re.compile(r"^(?:([^@:/\s]+)@)?(\[[^\]\s]+\]|[^@:/\s]+):(.+)$")
_WINDOWS_PATH = re.compile(r"^[A-Za-z]:[\\/]")
SSH_ON = [
" - Mac: System Settings > General > Sharing > turn on 'Remote Login'.",
" - Linux: run sudo apt install openssh-server and then sudo systemctl enable --now ssh",
" - Windows: Settings > Apps > Optional features > add 'OpenSSH Server', then in an Administrator",
" PowerShell run: Start-Service sshd; Set-Service sshd -StartupType Automatic",
]
class TargetError(Exception):
"""The target file could not be reached or changed. `why` says what happened, `advice` how to fix it."""
def __init__(self, why, advice=()):
super().__init__(why)
self.why = why
self.advice = list(advice)
class EnvLine:
"""One logical line of a .env file: a comment, a blank, an assignment (with the following lines of
a multi-line quoted value) or anything else, which is kept as it is."""
__slots__ = ("kind", "text", "eol", "head", "key", "value", "more")
def __init__(self, kind, text, eol, head="", key="", value=""):
self.kind, self.text, self.eol = kind, text, eol
self.head, self.key, self.value = head, key, value
self.more = [] # (text, line ending) of the lines that continue a multi-line value
def raw(self):
return self.text + self.eol + "".join(t + e for t, e in self.more)
def _text(data):
"""A .env file's bytes as text (odd bytes are kept as they are), without the byte-order mark."""
text = data.decode("utf-8", "surrogateescape")
return text[1:] if text.startswith("\ufeff") else text
def _split_lines(text):
"""(content, line ending) per line; the ending is '' only for a last line that has none."""
parts = text.split("\n")
out = []
for i, part in enumerate(parts):
if i == len(parts) - 1:
if part:
out.append((part, ""))
elif part.endswith("\r"):
out.append((part[:-1], "\r\n"))
else:
out.append((part, "\n"))
return out
def _closes(rest, quote):
"""True if `rest` (the text after an opening quote) contains the matching closing quote."""
i = 0
while i < len(rest):
if rest[i] == "\\" and quote != "`":
i += 2
continue
if rest[i] == quote:
return True
i += 1
return False
def scan_env(text):
"""Split .env text into EnvLine objects. A line whose first character (after spaces) is # is a comment
and is never taken for a key."""
lines = _split_lines(text)
out, i = [], 0
while i < len(lines):
content, eol = lines[i]
i += 1
stripped = content.strip(" \t")
if not stripped:
out.append(EnvLine("blank", content, eol))
continue
if stripped.startswith("#"):
out.append(EnvLine("comment", content, eol))
continue
m = _ASSIGN.match(content)
if not m:
out.append(EnvLine("other", content, eol))
continue
line = EnvLine("assign", content, eol, head=m.group(1) + m.group(2) + m.group(3),
key=m.group(2), value=m.group(4))
quote = line.value[:1]
if quote in ("'", '"', "`") and not _closes(line.value[1:], quote):
# a quoted value that goes on over the next lines, up to the one with the closing quote;
# without a closing quote anywhere it is taken as a one-line value
j = i
while j < len(lines) and not _closes(lines[j][0], quote):
j += 1
if j < len(lines):
line.more = lines[i:j + 1]
i = j + 1
out.append(line)
return out
def env_pairs(text):
"""The active KEY=VALUE lines of an instance's .env, in file order: [(key, value), ...]."""
return [(e.key, e.value) for e in scan_env(text) if e.kind == "assign"]
def merge_env(data, pairs):
"""The .env file `data` (bytes) with `pairs` put in: an existing key gets the new value, a missing key
is added at the end. Every other line stays byte for byte as it was.
Returns (new bytes, report); the report lists the keys replaced, added and already correct."""
if b"\x00" in data:
raise TargetError("the file is not plain text (it looks like it was saved as UTF-16).",
["If you want to fix it: open the file in an editor, save it again as UTF-8 and "
"run this again.", "Nothing was changed."])
entries = scan_env(_text(data))
wanted = dict(pairs)
endings = [x for e in entries for x in [e.eol] + [m[1] for m in e.more]]
eol = "\r\n" if endings.count("\r\n") > endings.count("\n") else "\n"
status, out = {}, []
for e in entries:
if e.kind == "assign" and e.key in wanted:
same = e.value == wanted[e.key] and not e.more
status[e.key] = "same" if same and status.get(e.key, "same") == "same" else "replaced"
out.append(e.raw() if same else e.head + wanted[e.key] + (e.more[-1][1] if e.more else e.eol))
else:
out.append(e.raw())
missing = [k for k in wanted if k not in status]
if missing:
if out and not out[-1].endswith("\n"):
out[-1] += eol
out += [f"{k}={wanted[k]}{eol}" for k in missing]
text = ("\ufeff" if data.startswith(b"\xef\xbb\xbf") else "") + "".join(out)
report = {
"replaced": [k for k in wanted if status.get(k) == "replaced"],
"same": [k for k in wanted if status.get(k) == "same"],
"added": missing,
"lines": sum(1 + len(e.more) for e in entries),
"untouched": sum(1 + len(e.more) for e in entries if not (e.kind == "assign" and e.key in wanted)),
"comments": sum(e.kind == "comment" for e in entries),
}
return text.encode("utf-8", "surrogateescape"), report
def _other_pairs(data, wanted):
"""Every active KEY=VALUE line that is not one of the instance's keys. Deliberately simple (one line at a
time, no multi-line values) so that it does not share the merge's own understanding of the file."""
out = []
for line in _text(data).split("\n"):
line = line.rstrip("\r")
m = _PAIR.match(line)
if m and not line.lstrip(" \t").startswith("#") and m.group(1) not in wanted:
out.append((m.group(1), m.group(2)))
return out
def _kept(entries, wanted):
"""The lines that are not the instance's, as text. The only change allowed to them is a line ending on the
very last line: a file without a final newline gets one when keys are added after it."""
kept = [e.raw() for e in entries if not (e.kind == "assign" and e.key in wanted)]
if kept:
kept[-1] = kept[-1].rstrip("\r\n")
return kept
def check_merge(before, after, pairs):
"""Compare the target file before the change and as it is now. Returns what is wrong (empty = all good):
nothing but the instance's keys may differ, and each of them must have exactly the instance's value."""
wanted = dict(pairs)
b, a = scan_env(_text(before)), scan_env(_text(after))
problems = []
if _kept(b, wanted) != _kept(a, wanted):
problems.append("lines that are not the instance's were changed, moved or removed")
if _other_pairs(before, wanted) != _other_pairs(after, wanted):
problems.append("a key that is not the instance's lost or changed its value")
for key, value in wanted.items():
found = [e for e in a if e.kind == "assign" and e.key == key]
if not found:
problems.append(f"{key} is missing")
elif any(e.value != value or e.more for e in found):
problems.append(f"{key} does not have the instance's value")
return problems
def parse_target(text, home=None):
"""Where the .env file is: '/path', '~/path' or 'user@host:path'.
Returns {'kind': 'local', 'path'} or {'kind': 'ssh', 'user', 'host', 'path'}."""
text = text.strip()
if not text:
raise ValueError("Nothing was typed.")
if text == "~" or text.startswith(("/", "./", "../", "~/")):
return {"kind": "local", "path": home + text[1:] if home and text.startswith("~") else text}
m = _REMOTE.match(text)
if not m:
return {"kind": "local", "path": text}
user, host, path = m.groups()
if len(host) == 1 and path[0] == "\\":
raise ValueError("That looks like a Windows path without the computer's name. "
"Write it like this: user@computer:C:/Users/you/app/.env")
if _WINDOWS_PATH.match(path):
path = "/" + path.replace("\\", "/")
return {"kind": "ssh", "user": user, "host": host, "path": path}
def invoking_user():
"""The passwd entry of the person who ran sudo, or None when supaspin runs directly as root."""
name = os.environ.get("SUDO_USER")
if os.geteuid() != 0 or not name or name == "root":
return None
try:
return pwd.getpwnam(name)
except KeyError:
return None
def private_tmpdir(owner):
"""A folder that only this run can read, in memory (/run) when possible: it holds the safety copy of the
target file while the change is made."""
base = "/run" if os.path.isdir("/run") and os.access("/run", os.W_OK) else None
d = tempfile.mkdtemp(prefix="supaspin-", dir=base)
if owner:
os.chown(d, owner.pw_uid, owner.pw_gid)
return d
def put_private(path, data, owner):
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "wb") as f:
f.write(data)
if owner:
os.chown(path, owner.pw_uid, owner.pw_gid)
def ssh_error(err, spec, me):
"""Turn what scp/ssh printed into a TargetError with advice in plain words."""
host, port, path = spec["host"], spec["port"], spec["path"]
user = spec.get("user") or me
hand = f" ssh -p {port} {user}@{host}"
low = err.lower()
first = next((l.strip() for l in err.splitlines() if l.strip()), "no message")
if ("identity file" in low and "not accessible" in low) or "load key" in low or "incorrect passphrase" in low:
return TargetError(f"the SSH key file cannot be used: {first}",
[f"If you want to fix it: check that the path is right, that {me} can read the file "
"and that only its owner may (chmod 600 <the key file>)."])
if "could not resolve hostname" in low:
advice = ["If you want to fix it: use the computer's IP address (for example 192.168.10.50) instead of its "
"name, or check the spelling.", "A name that works on your own computer may not be known to this server."]
if len(host) == 1:
advice.append(f"If you meant a Windows path such as {host}:/Users/..., write it as "
f"user@computer:{host}:/Users/...")
return TargetError(f"the name '{host}' is not known to this server.", advice)
if "remote host identification has changed" in low:
forget = host if port == 22 else f"[{host}]:{port}"
return TargetError(f"the computer at {host} identifies itself differently than last time.", [
"This happens when that computer was reinstalled or when a different computer now has that address. "
"It can also mean that somebody is pretending to be it.",
f"If you are sure it is the right computer, run this on this server (as {me}, without sudo) to forget "
"the old fingerprint, then try again:", f" ssh-keygen -R {forget}"])
if "host key verification failed" in low:
return TargetError(f"this server does not trust {host} yet (its fingerprint is not on the list of "
"known computers).", [
f"If you want to trust that computer, run this once on this server (as {me}, without sudo), check the "
"fingerprint it shows and answer yes:", hand, "Then run this again."])
if any(s in low for s in ("timed out", "no route to host", "network is unreachable")):
return TargetError(f"{host} did not answer on port {port} (no connection could be made).", [
"This usually means this server cannot reach that computer at all.",
"If you want to fix it, check these in this order:",
" 1. Is the computer switched on, awake and connected to the network?",
" 2. Is the address right? Addresses can change - use the computer's current one.",
" 3. Is SSH switched on there?", *SSH_ON,
f" 4. Is a firewall or your router blocking port {port}?",
"To try the connection yourself, run this on this server:", hand])
if "connection refused" in low:
return TargetError(f"{host} answered, but nothing is listening for SSH on port {port}.", [
"If you want to fix it: switch SSH on there, or use the right port number.", *SSH_ON,
"To try the connection yourself, run this on this server:", hand])
if "permission denied (" in low or "too many authentication failures" in low or "authentication failed" in low:
return TargetError(f"{host} refused the login for the user '{user}'.", [
"Either the user name or the password is wrong, or that computer does not know this server's key.",
"If you want to log in with a password: run this again in a terminal (not from a script) and type the "
"password when SSH asks for it.",
f"If you want this server to log in without a password: run this on this server (as {me}, without "
"sudo). It asks for the password once and installs the key:", f" ssh-copy-id -p {port} {user}@{host}",
"ssh-copy-id does not work for a Windows computer: copy the content of ~/.ssh/id_ed25519.pub from this "
"server into C:\\Users\\<user>\\.ssh\\authorized_keys there (for an administrator account: "
"C:\\ProgramData\\ssh\\administrators_authorized_keys).",
"If the user name is wrong, run this again with the right one: user@computer:/path/.env"])
if "no such file" in low:
return TargetError(f"the file {path} does not exist on {host}.",
["If you want to fix it: check the spelling and the folder. Nothing was created."])
if "permission denied" in low:
return TargetError(f"the login worked, but user '{user}' is not allowed to open or change {path} on {host}.",
["If you want to fix it: give that user write permission on the file (and read "
"permission on its folder), or use the user that owns the file."])
if "subsystem request failed" in low:
return TargetError(f"{host} accepted the login but does not allow file copying (SFTP).", [
"If you want to fix it: on that computer make sure the SSH server settings contain a line starting "
"with 'Subsystem sftp', then restart SSH (Linux: sudo systemctl restart ssh). Mac and Windows have it "
"on by default."])
return TargetError(f"the connection to {host} failed: {first}",
["To see more, try the connection by hand on this server:", hand])
class LocalTarget:
"""A .env file on this server, changed in place so that its owner and permissions stay as they are."""
def __init__(self, spec):
self.path = self.label = spec["path"]
def _error(self, e):
if isinstance(e, FileNotFoundError):
return TargetError(f"the file {self.path} does not exist on this server.",
["If you want to fix it: check the spelling and the folder. Nothing was created."])
if isinstance(e, PermissionError):
return TargetError(f"this server is not allowed to open {self.path}.")
return TargetError(f"{self.path} cannot be used: {e.strerror or e}.")
def read(self):
try:
with open(self.path, "rb") as f:
return f.read()
except OSError as e:
raise self._error(e)
def write(self, data):
try:
with open(self.path, "r+b") as f:
f.write(data)
f.truncate()
f.flush()
os.fsync(f.fileno())
except OSError as e:
raise self._error(e)
def close(self):
pass
class SshTarget:
"""A .env file on another computer, copied with scp. scp needs nothing on the other side except an SSH
server with file copying (SFTP), so this works for Mac, Linux and Windows computers. SSH runs as the person
who typed sudo: their keys and their list of known computers apply. One connection, opened first, is shared
by all copies, so a password is asked only once."""
def __init__(self, spec, tmp, owner):
self.spec, self.tmp, self.owner = spec, tmp, owner
self.me = owner.pw_name if owner else "root"
self.dest = f"{spec['user']}@{spec['host']}" if spec.get("user") else spec["host"]
self.label = f"{self.dest}:{spec['path']}"
self.sock = os.path.join(tmp, "c")
self.opts = ["-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=10", "-o", "ServerAliveCountMax=3",
"-o", f"Port={spec['port']}"]
if not sys.stdin.isatty():
self.opts += ["-o", "BatchMode=yes"] # never wait for a password nobody can type
if spec.get("key"):
self.opts += ["-i", spec["key"], "-o", "IdentitiesOnly=yes"] # -i (not -o) so ssh warns about a bad path
self.connected = False
def _as_owner(self, cmd):
return ["runuser", "-u", self.owner.pw_name, "--"] + cmd if self.owner else cmd
def _connect(self):
"""Open the one connection that all copies share. This is where SSH asks for a password or a key
passphrase, once per run. (scp cannot open a shared connection by itself, so ssh does it.)"""
if self.connected:
return
with tempfile.TemporaryFile("w+") as err:
p = subprocess.run(self._as_owner(["ssh", "-f", "-N", "-M", "-S", self.sock, *self.opts, self.dest]),
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=err)
err.seek(0)
text = err.read()
if p.returncode != 0:
raise ssh_error(text, self.spec, self.me)
self.connected = True
def _scp(self, src, dst):
self._connect()
p = subprocess.run(self._as_owner(["scp", "-o", f"ControlPath={self.sock}", *self.opts, src, dst]),
stdin=subprocess.DEVNULL, capture_output=True, text=True)
if p.returncode != 0:
raise ssh_error(p.stderr or p.stdout, self.spec, self.me)
def read(self):
dst = os.path.join(self.tmp, "download")
self._scp(f"{self.dest}:{self.spec['path']}", dst)
with open(dst, "rb") as f:
data = f.read()
os.remove(dst)
return data
def write(self, data):
src = os.path.join(self.tmp, "upload")
put_private(src, data, self.owner)
try:
self._scp(src, f"{self.dest}:{self.spec['path']}")
finally:
os.remove(src)
def close(self):
if os.path.exists(self.sock):
subprocess.run(self._as_owner(["ssh", "-o", f"ControlPath={self.sock}", "-O", "exit", self.dest]),
stdin=subprocess.DEVNULL, capture_output=True)
def open_target(spec, tmp, owner):
return LocalTarget(spec) if spec["kind"] == "local" else SshTarget(spec, tmp, owner)
def undo_write(target, original):
"""After a failed write: 'unchanged' (the file never changed), 'restored' (put back and confirmed) or
'failed' (the file may be damaged and could not be put back)."""
try:
if target.read() == original:
return "unchanged"
target.write(original)
return "restored" if target.read() == original else "failed"
except TargetError:
return "failed"
def _print_changes(log, rep, dry):
names = ("Would replace", "Would add", "Already correct") if dry else ("Replaced", "Added", "Already correct")
for name, keys in zip(names, (rep["replaced"], rep["added"], rep["same"])):
log(f" {name} ({len(keys)}): {', '.join(keys) if keys else '(none)'}")
log(f" Untouched: {rep['untouched']} other lines (other keys, comments, blank lines) "
f"{'stay' if dry else 'are'} exactly as they {'are' if dry else 'were'}.")
def push_env(cfg, n, spec, log, dry_run=False):
"""Put instance n's .env values into the target file and report what happened. Before anything is
written, the target is copied to a private temporary folder; after writing, the file is read back and
compared with that copy. If anything is off, the copy is written back. The copy is deleted at the end,
except when it could not be written back (then it is the only good copy). Returns 0 or 1."""
who = f"instance {n} ({cfg['slots'].get(n, '?')})"
try:
pairs = env_pairs(read_env(cfg, n))
except FileNotFoundError:
log(f"{RED}FAILED{OFF} {who} has no .env file ({inst_dir(cfg, n)}/.env), so there is nothing to copy.")
return 1
owner = invoking_user()
tmp = private_tmpdir(owner)
target, written, keep = None, False, False
try:
try:
target = open_target(spec, tmp, owner)
log(f"Reading {target.label} ...")
original = target.read()
put_private(os.path.join(tmp, "original.env"), original, None)
new, rep = merge_env(original, pairs)
log(f" ok - {rep['lines']} lines, {rep['comments']} commented-out lines are ignored")
log(f"Working out the changes: {len(rep['replaced'])} to replace, {len(rep['added'])} to add, "
f"{len(rep['same'])} already correct")
if dry_run:
log(f"{BOLD}DRY RUN{OFF} {who} -> {target.label} Nothing was written.")
_print_changes(log, rep, dry=True)
return 0
if new == original:
log(f"{BOLD}SUCCESS{OFF} {who} -> {target.label}")
log(" Nothing to change: every value in the file was already correct. The file was not touched.")
return 0
log("Writing the file ...")
written = True # from here on the file may be damaged, even if the write itself fails
target.write(new)
log("Reading it back and checking that nothing else changed ...")
back = target.read()
problems = check_merge(original, back, pairs)
if not problems and back != new:
problems = ["the file that arrived is not the one that was sent"]
if problems:
raise TargetError("the check after writing failed: " + "; ".join(problems) + ".")
except (TargetError, KeyboardInterrupt) as e:
err = e if isinstance(e, TargetError) else TargetError("it was interrupted (Ctrl+C).")
state = undo_write(target, original) if written else "unchanged"
keep = state == "failed"
log(f"{RED}FAILED{OFF} {who} was NOT written to {target.label}")
log(f" Why: {err.why}")
if state == "unchanged":
log(" Your file was not changed.")
elif state == "restored":
log(" Your file was restored to exactly how it was. Nothing was lost.")
else:
log(f" {RED}WARNING: the file may be damaged and could not be put back.{OFF}")
log(f" An untouched copy of the original is kept on this server: {tmp}/original.env")
log(f" Copy it back to {target.label} by hand, then delete that folder: sudo rm -r {tmp}")
for line in err.advice:
log(f" {line}" if line else "")
return 1
log(f"{BOLD}SUCCESS{OFF} {who} -> {target.label}")
_print_changes(log, rep, dry=False)
return 0
finally:
if target is not None:
target.close()
if not keep:
shutil.rmtree(tmp, ignore_errors=True)
def ask_target(user):
"""Ask where the .env file is and how to log in there. Returns a target spec, or None."""
print("Which .env file should receive the values? Type where it is:")
print(" on this server: /home/igor/app/.env")
print(" on a Mac or Linux computer: igor@192.168.10.50:/Users/igor/app/.env")
print(" on a Windows computer: igor@192.168.10.50:C:/Users/igor/app/.env")
print("The file must already exist. Only the instance's own keys are changed; everything else stays.\n")
while True:
text = ask("File (Enter to cancel)")
if not text:
return None
try:
spec = parse_target(text, user.pw_dir if user else None)
break
except ValueError as e:
print(f"{RED}{e}{OFF}")
if spec["kind"] == "ssh":
if not spec["user"]:
spec["user"] = ask("SSH user on that computer", user.pw_name if user else "root")
try:
spec["port"] = int(ask("SSH port", "22"))
except ValueError:
print(f"{RED}The port must be a number.{OFF}")
return None
spec["key"] = ask("SSH key file (press Enter to use your normal keys, or to type a password when asked)") or None
return spec
def push_env_command(cfg, args):
ap = argparse.ArgumentParser(prog="supaspin push-env", description="Put an instance's .env values into "
"another .env file: existing keys get the new value, missing keys are added.")
ap.add_argument("number", type=int, help="instance number (see: supaspin list)")
ap.add_argument("target", nargs="?", help="/path/.env on this server, or user@host:/path/.env on another "
"computer (Windows: user@host:C:/path/.env). Put it in quotes. Asked for if left out.")
ap.add_argument("--port", type=int, default=22, help="SSH port of the other computer (default 22)")
ap.add_argument("--key", help="SSH key file to log in with (default: your normal keys, or a password)")
ap.add_argument("--dry-run", action="store_true", help="show what would change, write nothing")
a = ap.parse_args(args)
if a.number not in existing_numbers(cfg):
print(f"{RED}Instance {a.number} does not exist.{OFF} See: sudo supaspin list", file=sys.stderr)
return 1
user = invoking_user()
if a.target:
try:
spec = parse_target(a.target, user.pw_dir if user else None)
except ValueError as e:
print(f"{RED}{e}{OFF}", file=sys.stderr)
return 1
if spec["kind"] == "ssh":
spec.update(port=a.port, key=a.key)
elif sys.stdin.isatty():
spec = ask_target(user)
if spec is None:
return 1
else:
print(f"{RED}Say where the file is: supaspin push-env {a.number} user@host:/path/.env{OFF}", file=sys.stderr)
return 1
return push_env(cfg, a.number, spec, log_line, a.dry_run)
# --------------------------------------------------------------------- ui --
def log_line(msg, live=False):
@@ -912,13 +1543,17 @@ def print_list(cfg):
for r in rows:
print(f"{BOLD}{r['n']}{OFF} {r['host']} pods {r['pods']} memory {r['memory']}")
print(f" disk {r['disk']}\n env {r['env']}")
for line in tunnel_guide(cfg, r["n"]):
print(f" {line}" if line else "")
print()
def main_menu(cfg, log):
while True:
i = menu(f"supaspin {VERSION}", ["Create instance", "List instances", "Show instance",
"Delete instance", "Mail settings", "Settings", "Quit"])
if i is None or i == 6:
"Delete instance", "Copy instance values into another .env file",
"Mail settings", "Settings", "Quit"])
if i is None or i == 7:
return
print()
try:
@@ -936,9 +1571,15 @@ def main_menu(cfg, log):
if n and confirm_number(n, "deleting it"):
delete_instance(cfg, n, log)
elif i == 4:
n = pick_instance(cfg, "Copy which instance's values?")
spec = ask_target(invoking_user()) if n else None
if spec:
print()
push_env(cfg, n, spec, log)
elif i == 5:
cfg["mail"] = mail_dialog(cfg)
save_config(cfg)
elif i == 5:
elif i == 6:
cfg = setup_dialog(cfg, log)
except Exception as e:
print(f"{RED}{e}{OFF}")
@@ -959,7 +1600,7 @@ def main(argv):
cmd = argv[1] if len(argv) > 1 else "menu"
cfg = load_config()
if cmd in ("--version", "version"):
print(VERSION)
print(f"supaspin {VERSION}")
return 0
try:
if cfg is None or cmd == "setup":
@@ -987,11 +1628,11 @@ def main(argv):
elif cmd == "delete" and n:
if "--yes" in argv or confirm_number(n, "deleting it"):
delete_instance(cfg, n, log_line)
elif cmd == "push-env":
return push_env_command(cfg, argv[2:])
elif cmd == "mail":
cfg["mail"] = mail_dialog(cfg)
save_config(cfg)
elif cmd in ("help", "-h", "--help"):
print(__doc__)
else:
print(__doc__)
return 2
tests/test_supaspin.py
+603
View File
@@ -4,9 +4,18 @@ import hashlib
import hmac
import importlib.machinery
import importlib.util
import collections
import contextlib
import io
import json
import os
import random
import re
import shutil
import sys
import tempfile
import unittest
from unittest import mock
import yaml
@@ -323,5 +332,599 @@ class Manifests(unittest.TestCase):
self.assertIn("192.168.10.30:5441", text)
class Version(unittest.TestCase):
def test_version_file_code_and_readme_agree(self):
with open(os.path.join(ROOT, "VERSION")) as f:
version = f.read().strip()
self.assertRegex(version, r"^\d+\.\d+\.\d+$")
self.assertEqual(sp.VERSION, version)
with open(os.path.join(ROOT, "README.md")) as f:
self.assertEqual(f.read().splitlines()[2], f"Version {version}")
def test_version_command_names_the_app(self):
out = io.StringIO()
with mock.patch.object(sp.os, "geteuid", return_value=0), mock.patch.object(sp, "load_config", return_value=CFG), \
contextlib.redirect_stdout(out):
self.assertEqual(sp.main(["supaspin", "--version"]), 0)
self.assertEqual(out.getvalue().strip(), f"supaspin {sp.VERSION}")
class TunnelGuide(unittest.TestCase):
def test_list_guide_has_the_three_commands_for_this_instance(self):
text = "\n".join(sp.tunnel_guide(CFG, 2))
self.assertIn("ssh -N -p 2222 -L 5442:192.168.10.30:5442 -L 3002:192.168.10.30:3002 igor@203.0.113.5", text)
self.assertIn("ssh -f -N -M -S /tmp/supaspin-2.sock -o ExitOnForwardFailure=yes -p 2222 "
"-L 5442:192.168.10.30:5442 -L 3002:192.168.10.30:3002 igor@203.0.113.5", text)
self.assertIn("ssh -S /tmp/supaspin-2.sock -O exit igor@203.0.113.5", text)
self.assertIn("If you want to open the tunnel", text)
self.assertIn("If you want the tunnel to keep running in the background", text)
self.assertIn("If you no longer need the background tunnel", text)
def test_windows_limit_is_written_down(self):
for detailed in (False, True):
text = "\n".join(sp.tunnel_guide(CFG, 1, detailed))
self.assertIn("do NOT work on Windows", text)
self.assertIn("ONLY the first command below", text)
self.assertIn("(Mac or Linux only)", text)
self.assertIn("do NOT work on Windows", sp.tunnel_help(CFG, 1))
def test_list_guide_is_short_and_points_to_the_detailed_help(self):
text = "\n".join(sp.tunnel_guide(CFG, 1))
self.assertIn("sudo supaspin tunnel 1", text)
self.assertNotIn("pkill", text)
self.assertNotIn("-O check", text)
def test_detailed_guide_adds_check_and_leftover_tunnel(self):
text = "\n".join(sp.tunnel_guide(CFG, 1, detailed=True))
self.assertIn("ssh -S /tmp/supaspin-1.sock -O check igor@203.0.113.5", text)
self.assertIn('pkill -f "[s]upaspin-1.sock"', text)
self.assertIn("stops the command from also matching", text)
def test_without_public_ssh_host_it_says_how_to_set_it(self):
text = "\n".join(sp.tunnel_guide(dict(CFG, ssh_host=""), 1))
self.assertIn("Public SSH host for tunnels", text)
self.assertNotIn("ssh -N", text)
def test_list_prints_the_guide_under_each_instance(self):
rows = [{"n": 1, "host": "supa1.example.com", "pods": "12/12", "memory": "1/2 MiB", "disk": "db 1/2M",
"env": "/x/1/.env"}]
out = io.StringIO()
with mock.patch.object(sp, "list_rows", return_value=rows), contextlib.redirect_stdout(out):
sp.print_list(CFG)
text = out.getvalue()
self.assertIn("supa1.example.com", text)
self.assertIn("ssh -N -p 2222 -L 5441:192.168.10.30:5441", text)
self.assertIn("ssh -S /tmp/supaspin-1.sock -O exit", text)
self.assertIn("do NOT work on Windows", text)
class TargetParsing(unittest.TestCase):
def test_local_paths(self):
self.assertEqual(sp.parse_target("/srv/app/.env"), {"kind": "local", "path": "/srv/app/.env"})
self.assertEqual(sp.parse_target("./app/.env")["kind"], "local")
self.assertEqual(sp.parse_target("~/app/.env", "/home/igor")["path"], "/home/igor/app/.env")
self.assertEqual(sp.parse_target("app/.env"), {"kind": "local", "path": "app/.env"})
def test_remote_paths(self):
self.assertEqual(sp.parse_target("igor@192.168.10.50:/Users/igor/app/.env"),
{"kind": "ssh", "user": "igor", "host": "192.168.10.50", "path": "/Users/igor/app/.env"})
self.assertEqual(sp.parse_target("mac.local:app/.env")["user"], None)
self.assertEqual(sp.parse_target("u@[fe80::1]:/x/.env")["host"], "[fe80::1]")
def test_windows_paths_are_converted(self):
for typed in ("igor@pc:C:\\Users\\igor\\app\\.env", "igor@pc:C:/Users/igor/app/.env"):
self.assertEqual(sp.parse_target(typed)["path"], "/C:/Users/igor/app/.env")
def test_windows_path_without_computer_is_refused(self):
with self.assertRaises(ValueError) as cm:
sp.parse_target("C:\\Users\\igor\\.env")
self.assertIn("Windows path without the computer's name", str(cm.exception))
def test_nothing_typed(self):
with self.assertRaises(ValueError):
sp.parse_target(" ")
PAIRS = [("A", "1"), ("B", "2"), ("C", "3")]
class MergeEnv(unittest.TestCase):
def merge(self, data, pairs=PAIRS):
return sp.merge_env(data, pairs)
def test_replaces_existing_and_adds_missing_at_the_end(self):
new, rep = self.merge(b"A=old\nX=keep\n")
self.assertEqual(new, b"A=1\nX=keep\nB=2\nC=3\n")
self.assertEqual((rep["replaced"], rep["added"], rep["same"]), (["A"], ["B", "C"], []))
def test_commented_out_keys_are_not_taken_into_consideration(self):
data = b"# A=old\n#B=x\n # C=3\nX=1\n"
new, rep = self.merge(data)
self.assertEqual(new, data + b"A=1\nB=2\nC=3\n")
self.assertEqual(rep["added"], ["A", "B", "C"])
self.assertEqual(rep["comments"], 3)
def test_export_prefix_and_spacing_are_kept(self):
new, _ = self.merge(b"export A=old\nB = old\n C=old\n")
self.assertEqual(new, b"export A=1\nB = 2\n C=3\n")
def test_key_twice_is_replaced_in_both_places(self):
new, rep = self.merge(b"A=1\nA=2\n", [("A", "1")])
self.assertEqual(new, b"A=1\nA=1\n")
self.assertEqual(rep["replaced"], ["A"])
def test_similar_key_names_are_not_touched(self):
data = b"JWT_SECRET_OLD=1\nJWT_SECRETX=2\njwt_secret=3\nJWT_SECRET=old\n"
new, _ = self.merge(data, [("JWT_SECRET", "s")])
self.assertEqual(new, b"JWT_SECRET_OLD=1\nJWT_SECRETX=2\njwt_secret=3\nJWT_SECRET=s\n")
def test_windows_line_endings(self):
new, _ = self.merge(b"A=old\r\nX=1\r\n")
self.assertEqual(new, b"A=1\r\nX=1\r\nB=2\r\nC=3\r\n")
def test_missing_final_newline(self):
self.assertEqual(self.merge(b"X=1")[0], b"X=1\nA=1\nB=2\nC=3\n")
self.assertEqual(self.merge(b"A=old", [("A", "1")])[0], b"A=1") # nothing appended: stays as it was
def test_byte_order_mark_is_kept(self):
new, _ = self.merge(b"\xef\xbb\xbfA=old\n")
self.assertTrue(new.startswith(b"\xef\xbb\xbfA=1\n"))
def test_empty_file(self):
self.assertEqual(self.merge(b"")[0], b"A=1\nB=2\nC=3\n")
def test_multi_line_quoted_value_is_not_looked_into(self):
data = b'KEY="line1\nA=inside\nline3"\nA=old\n'
new, rep = self.merge(data)
self.assertEqual(new, b'KEY="line1\nA=inside\nline3"\nA=1\nB=2\nC=3\n')
self.assertEqual(rep["untouched"], 3)
def test_multi_line_value_of_an_instance_key_is_replaced_whole(self):
new, _ = self.merge(b'A="x\ny"\nX=1\n', [("A", "1")])
self.assertEqual(new, b"A=1\nX=1\n")
def test_unclosed_quote_does_not_swallow_the_rest(self):
new, _ = self.merge(b'X="oops\nA=old\n', [("A", "1")])
self.assertEqual(new, b'X="oops\nA=1\n')
def test_values_are_copied_verbatim(self):
value = "postgresql://postgres:pw@localhost:5441/postgres?a=b#c d"
new, _ = self.merge(b"", [("DATABASE_URL", value)])
self.assertEqual(new, f"DATABASE_URL={value}\n".encode())
def test_empty_instance_value_is_written_empty(self):
self.assertEqual(self.merge(b"U=real\n", [("U", "")])[0], b"U=\n")
def test_text_after_the_old_value_goes_with_it(self):
self.assertEqual(self.merge(b"A=old # note\n", [("A", "1")])[0], b"A=1\n")
def test_odd_bytes_elsewhere_survive(self):
data = b"X=caf\xe9\n# \xff\xfe\nA=old\n"
new, _ = self.merge(data, [("A", "1")])
self.assertEqual(new, b"X=caf\xe9\n# \xff\xfe\nA=1\n")
def test_second_run_changes_nothing(self):
first, _ = self.merge(b"A=old\nX=1\n")
second, rep = self.merge(first)
self.assertEqual(second, first)
self.assertEqual((rep["replaced"], rep["added"], rep["same"]), ([], [], ["A", "B", "C"]))
def test_utf16_file_is_refused(self):
with self.assertRaises(sp.TargetError) as cm:
self.merge("A=1\n".encode("utf-16"))
self.assertIn("UTF-16", cm.exception.why)
def test_report_counts(self):
_, rep = self.merge(b"# c\n\nX=1\nA=old\n")
self.assertEqual((rep["lines"], rep["untouched"], rep["comments"]), (4, 3, 1))
def test_instance_env_pairs_skip_comments(self):
pairs = sp.env_pairs(sp.render_env(CFG, 1, sp.generate_secrets(), MAIL))
self.assertEqual(len(pairs), 24)
self.assertEqual(pairs[0], ("SUPABASE_URL", "https://supa1.example.com"))
class CheckMerge(unittest.TestCase):
BEFORE = b"# note\nX=1\nA=old\n\nY=2\n"
def test_a_good_merge_passes(self):
after, _ = sp.merge_env(self.BEFORE, PAIRS)
self.assertEqual(sp.check_merge(self.BEFORE, after, PAIRS), [])
def test_a_lost_unrelated_key_is_caught(self):
after, _ = sp.merge_env(self.BEFORE, PAIRS)
problems = sp.check_merge(self.BEFORE, after.replace(b"Y=2\n", b""), PAIRS)
self.assertTrue(problems)
def test_a_changed_unrelated_value_is_caught(self):
after, _ = sp.merge_env(self.BEFORE, PAIRS)
self.assertTrue(sp.check_merge(self.BEFORE, after.replace(b"X=1", b"X=9"), PAIRS))
def test_a_missing_or_wrong_instance_value_is_caught(self):
after, _ = sp.merge_env(self.BEFORE, PAIRS)
self.assertIn("B is missing", sp.check_merge(self.BEFORE, after.replace(b"B=2\n", b""), PAIRS))
self.assertIn("A does not have the instance's value",
sp.check_merge(self.BEFORE, after.replace(b"A=1", b"A=7"), PAIRS))
def test_random_files_never_lose_an_unrelated_key(self):
rnd = random.Random(7)
pool = [b"# A=old", b"#B=c", b" # C=3", b"", b"X=1", b"Y=two words", b"export Z=3", b"A=old",
b"B = old", b"export C=old", b'Q="multi', b"A=inside", b'end"', b"JWT=1", b"W='a b'", b"e=1"]
for _ in range(300):
lines = [rnd.choice(pool) for _ in range(rnd.randint(0, 12))]
eol = rnd.choice([b"\n", b"\r\n"])
data = eol.join(lines) + rnd.choice([b"", eol])
new, _ = sp.merge_env(data, PAIRS)
self.assertEqual(sp.check_merge(data, new, PAIRS), [], data)
again, rep = sp.merge_env(new, PAIRS)
self.assertEqual(again, new)
self.assertEqual(rep["replaced"] + rep["added"], [])
class FakeTarget:
"""A file that lives in memory, for testing what push_env does when things go wrong."""
def __init__(self, data, mode="ok"):
self.data, self.mode, self.label, self.dead, self.closed = data, mode, "fake:/app/.env", False, False
def read(self):
if self.dead:
raise sp.TargetError("the connection is gone.")
return self.data
def write(self, data):
if self.dead:
raise sp.TargetError("the connection is gone.")
if self.mode == "refuse":
raise sp.TargetError("not allowed to write.")
if self.mode == "drop-a-line": # goes wrong once; the write that puts the original back is fine
data, self.mode = data.replace(b"KEEP=1\n", b""), "ok"
elif self.mode == "die-half-way":
self.data, self.dead = data[:5], True
raise sp.TargetError("the connection dropped while writing.")
self.data = data
def close(self):
self.closed = True
class PushEnv(unittest.TestCase):
def setUp(self):
self.dir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.dir, True)
self.cfg = dict(CFG, storage_root=self.dir)
os.makedirs(f"{self.dir}/1")
self.secret = "S3cr3tValue"
with open(f"{self.dir}/1/.env", "w") as f:
f.write(f"# instance 1\nSUPABASE_URL=https://supa1.example.com\nJWT_SECRET={self.secret}\n")
self.tmps = []
real = sp.private_tmpdir
def tracked(owner):
d = real(owner)
self.tmps.append(d)
return d
for patcher in (mock.patch.object(sp, "private_tmpdir", tracked),
mock.patch.object(sp, "invoking_user", return_value=None)):
patcher.start()
self.addCleanup(patcher.stop)
self.addCleanup(lambda: [shutil.rmtree(d, True) for d in self.tmps])
self.out = []
def run_push(self, spec, dry_run=False):
return sp.push_env(self.cfg, 1, spec, self.out.append, dry_run)
def text(self):
return re.sub(r"\x1b\[[0-9;]*m", "", "\n".join(self.out))
def local_target(self, content):
path = f"{self.dir}/target.env"
with open(path, "wb") as f:
f.write(content)
return path
def fake(self, data, mode="ok"):
target = FakeTarget(data, mode)
patcher = mock.patch.object(sp, "open_target", return_value=target)
patcher.start()
self.addCleanup(patcher.stop)
return target
def test_success_changes_only_the_instances_keys_and_reports_names_not_values(self):
original = b"KEEP=1\n# JWT_SECRET=commented\nJWT_SECRET=old\nOTHER=x y\n"
path = self.local_target(original)
self.assertEqual(self.run_push({"kind": "local", "path": path}), 0)
with open(path, "rb") as f:
self.assertEqual(f.read(), b"KEEP=1\n# JWT_SECRET=commented\nJWT_SECRET=" + self.secret.encode() +
b"\nOTHER=x y\nSUPABASE_URL=https://supa1.example.com\n")
text = self.text()
self.assertIn("SUCCESS instance 1 (supa1.example.com)", text)
self.assertIn("Replaced (1): JWT_SECRET", text)
self.assertIn("Added (1): SUPABASE_URL", text)
self.assertNotIn(self.secret, text)
self.assertFalse(any(os.path.exists(d) for d in self.tmps), "temporary copy must be deleted")
def test_dry_run_writes_nothing(self):
path = self.local_target(b"KEEP=1\n")
self.assertEqual(self.run_push({"kind": "local", "path": path}, dry_run=True), 0)
with open(path, "rb") as f:
self.assertEqual(f.read(), b"KEEP=1\n")
self.assertIn("DRY RUN", self.text())
self.assertIn("Would add (2)", self.text())
def test_nothing_to_change_does_not_touch_the_file(self):
path = self.local_target(b"SUPABASE_URL=https://supa1.example.com\nJWT_SECRET=" + self.secret.encode() + b"\n")
before = os.stat(path).st_mtime_ns
self.assertEqual(self.run_push({"kind": "local", "path": path}), 0)
self.assertEqual(os.stat(path).st_mtime_ns, before)
self.assertIn("Nothing to change", self.text())
def test_missing_target_file_fails_and_is_not_created(self):
path = f"{self.dir}/nope.env"
self.assertEqual(self.run_push({"kind": "local", "path": path}), 1)
self.assertFalse(os.path.exists(path))
self.assertIn("FAILED", self.text())
self.assertIn("does not exist", self.text())
self.assertIn("Your file was not changed.", self.text())
def test_instance_without_env_fails(self):
os.remove(f"{self.dir}/1/.env")
self.assertEqual(self.run_push({"kind": "local", "path": "/x"}), 1)
self.assertIn("has no .env file", self.text())
def test_refused_write_leaves_the_file_alone(self):
target = self.fake(b"KEEP=1\n", "refuse")
self.assertEqual(self.run_push({"kind": "ssh"}), 1)
self.assertEqual(target.data, b"KEEP=1\n")
self.assertIn("Your file was not changed.", self.text())
self.assertTrue(target.closed)
self.assertFalse(any(os.path.exists(d) for d in self.tmps))
def test_a_write_that_loses_a_line_is_undone(self):
original = b"KEEP=1\nA=2\n"
target = self.fake(original, "drop-a-line")
self.assertEqual(self.run_push({"kind": "ssh"}), 1)
self.assertEqual(target.data, original)
self.assertIn("the check after writing failed", self.text())
self.assertIn("restored to exactly how it was", self.text())
self.assertFalse(any(os.path.exists(d) for d in self.tmps))
def test_if_the_file_cannot_be_put_back_the_copy_is_kept(self):
original = b"KEEP=1\nA=2\n"
target = self.fake(original, "die-half-way")
self.assertEqual(self.run_push({"kind": "ssh"}), 1)
text = self.text()
self.assertIn("WARNING: the file may be damaged", text)
self.assertIn("connection dropped while writing", text)
kept = [d for d in self.tmps if os.path.exists(d)]
self.assertEqual(len(kept), 1)
self.assertIn(kept[0] + "/original.env", text)
with open(kept[0] + "/original.env", "rb") as f:
self.assertEqual(f.read(), original)
self.assertNotEqual(target.data, original)
def test_advice_is_printed_with_the_failure(self):
with mock.patch.object(sp, "open_target", side_effect=lambda *a: FakeTarget(b"\xff\x00")):
self.assertEqual(self.run_push({"kind": "ssh"}), 1)
self.assertIn("save it again as UTF-8", self.text())
def test_local_write_keeps_mode_and_file_identity(self):
path = self.local_target(b"KEEP=1\nJWT_SECRET=a long old value that is longer than the new one\n")
os.chmod(path, 0o640)
inode = os.stat(path).st_ino
self.assertEqual(self.run_push({"kind": "local", "path": path}), 0)
st = os.stat(path)
self.assertEqual((st.st_ino, st.st_mode & 0o777), (inode, 0o640))
class SshTargetCommands(unittest.TestCase):
SPEC = {"kind": "ssh", "user": "igor", "host": "pc.lan", "port": 2200, "path": "/C:/app/.env", "key": "/k/id"}
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.tmp, True)
self.calls = []
def fake_run(self, content=b"A=1\n", master_fails="", scp_fails=""):
def run(cmd, **kw):
self.calls.append(cmd)
result = (0, "")
if "scp" in cmd:
if scp_fails:
result = (1, scp_fails)
elif "download" in cmd[-1]:
with open(cmd[-1], "wb") as f:
f.write(content)
elif "-M" in cmd and master_fails:
kw["stderr"].write(master_fails) # the master's messages go to a file
result = (255, "")
return subprocess_result(*result)
return mock.patch.object(sp.subprocess, "run", run)
def scp_call(self):
return next(c for c in self.calls if "scp" in c)
def test_one_shared_connection_is_opened_first_and_used_by_every_copy(self):
target = sp.SshTarget(self.SPEC, self.tmp, None)
with self.fake_run(b"A=1\n"):
self.assertEqual(target.read(), b"A=1\n")
target.write(b"A=2\n")
target.read()
masters = [c for c in self.calls if "-M" in c]
self.assertEqual(len(masters), 1)
self.assertEqual(masters[0][:4], ["ssh", "-f", "-N", "-M"])
self.assertEqual(masters[0][masters[0].index("-S") + 1], f"{self.tmp}/c")
self.assertEqual(masters[0][-1], "igor@pc.lan")
self.assertIs(self.calls[0], masters[0])
for scp in (c for c in self.calls if "scp" in c):
self.assertIn(f"ControlPath={self.tmp}/c", scp)
def test_read_copies_the_file_with_scp(self):
target = sp.SshTarget(self.SPEC, self.tmp, None)
with self.fake_run(b"A=1\n"):
target.read()
for cmd in self.calls:
self.assertIn("Port=2200", cmd)
self.assertEqual(cmd[cmd.index("-i") + 1], "/k/id")
self.assertIn("IdentitiesOnly=yes", cmd)
self.assertEqual(self.scp_call()[-2], "igor@pc.lan:/C:/app/.env")
def test_write_uploads_and_removes_its_temporary_file(self):
target = sp.SshTarget(self.SPEC, self.tmp, None)
with self.fake_run():
target.write(b"A=2\n")
cmd = self.scp_call()
self.assertEqual(cmd[-1], "igor@pc.lan:/C:/app/.env")
self.assertFalse(os.path.exists(cmd[-2]))
def test_ssh_runs_as_the_person_who_typed_sudo(self):
owner = collections.namedtuple("Owner", "pw_name pw_uid pw_gid")("igor", os.getuid(), os.getgid())
target = sp.SshTarget(self.SPEC, self.tmp, owner)
with self.fake_run():
target.read()
self.assertEqual(len(self.calls), 2)
for cmd in self.calls:
self.assertEqual(cmd[:4], ["runuser", "-u", "igor", "--"])
def test_without_a_terminal_it_never_waits_for_a_password(self):
with mock.patch("sys.stdin") as stdin:
stdin.isatty.return_value = False
self.assertIn("BatchMode=yes", sp.SshTarget(self.SPEC, self.tmp, None).opts)
stdin.isatty.return_value = True
self.assertNotIn("BatchMode=yes", sp.SshTarget(self.SPEC, self.tmp, None).opts)
def test_a_failed_login_becomes_advice(self):
target = sp.SshTarget(self.SPEC, self.tmp, None)
with self.fake_run(master_fails="ssh: connect to host pc.lan port 2200: Connection refused"):
with self.assertRaises(sp.TargetError) as cm:
target.read()
self.assertIn("nothing is listening for SSH on port 2200", cm.exception.why)
self.assertFalse(any("scp" in c for c in self.calls), "no copy is tried without a connection")
def test_a_failed_copy_becomes_advice(self):
target = sp.SshTarget(self.SPEC, self.tmp, None)
with self.fake_run(scp_fails="scp: /C:/app/.env: No such file or directory"):
with self.assertRaises(sp.TargetError) as cm:
target.read()
self.assertIn("does not exist on pc.lan", cm.exception.why)
def test_close_asks_the_shared_connection_to_quit(self):
target = sp.SshTarget(self.SPEC, self.tmp, None)
open(target.sock, "w").close()
with self.fake_run():
target.close()
self.assertEqual(self.calls[0][-3:], ["-O", "exit", "igor@pc.lan"])
self.assertIn(f"ControlPath={target.sock}", self.calls[0])
def subprocess_result(returncode, stderr):
return collections.namedtuple("R", "returncode stdout stderr")(returncode, "", stderr)
class SshErrors(unittest.TestCase):
SPEC = {"user": "igor", "host": "pc.lan", "port": 22, "path": "/app/.env", "key": None}
def why(self, err, **spec):
e = sp.ssh_error(err, dict(self.SPEC, **spec), "igor")
return e.why, "\n".join(e.advice)
def test_unknown_name(self):
why, advice = self.why("ssh: Could not resolve hostname pc.lan: Name or service not known")
self.assertIn("'pc.lan' is not known", why)
self.assertIn("IP address", advice)
def test_single_letter_host_hints_at_windows_paths(self):
_, advice = self.why("ssh: Could not resolve hostname C: Name or service not known", host="C")
self.assertIn("user@computer:C:/Users/", advice)
def test_timeout(self):
why, advice = self.why("ssh: connect to host pc.lan port 22: Connection timed out")
self.assertIn("did not answer on port 22", why)
self.assertIn("switched on", advice)
self.assertIn("ssh -p 22 igor@pc.lan", advice)
def test_refused_explains_how_to_switch_ssh_on_everywhere(self):
why, advice = self.why("ssh: connect to host pc.lan port 22: Connection refused")
self.assertIn("nothing is listening", why)
for word in ("Remote Login", "openssh-server", "OpenSSH Server"):
self.assertIn(word, advice)
def test_login_refused(self):
why, advice = self.why("igor@pc.lan: Permission denied (publickey,password).")
self.assertIn("refused the login for the user 'igor'", why)
self.assertIn("ssh-copy-id -p 22 igor@pc.lan", advice)
self.assertIn("administrators_authorized_keys", advice)
def test_unknown_computer_and_changed_computer(self):
why, advice = self.why("Host key verification failed.")
self.assertIn("does not trust pc.lan yet", why)
self.assertIn("answer yes", advice)
why, advice = self.why("@@@ WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED! @@@\nHost key verification failed.",
port=2222)
self.assertIn("identifies itself differently", why)
self.assertIn('ssh-keygen -R [pc.lan]:2222', advice)
def test_missing_file_and_protected_file(self):
why, _ = self.why("scp: /app/.env: No such file or directory")
self.assertIn("/app/.env does not exist on pc.lan", why)
why, advice = self.why('scp: dest open "/app/.env": Permission denied')
self.assertIn("not allowed to open or change /app/.env", why)
self.assertIn("write permission", advice)
def test_sftp_switched_off(self):
why, advice = self.why("subsystem request failed on channel 0\nscp: Connection closed")
self.assertIn("does not allow file copying (SFTP)", why)
self.assertIn("Subsystem sftp", advice)
def test_bad_key_file(self):
why, _ = self.why("Warning: Identity file /k/id not accessible: No such file or directory.\n"
"igor@pc.lan: Permission denied (publickey).", key="/k/id")
self.assertIn("SSH key file cannot be used", why)
def test_anything_else_shows_the_message_and_how_to_try_by_hand(self):
why, advice = self.why("kex_exchange_identification: read: Connection reset by peer")
self.assertIn("Connection reset by peer", why)
self.assertIn("ssh -p 22 igor@pc.lan", advice)
class PushEnvCommand(unittest.TestCase):
def run_cmd(self, args, numbers=(1, 2), tty=False):
calls = []
with mock.patch.object(sp, "existing_numbers", return_value=set(numbers)), \
mock.patch.object(sp, "invoking_user", return_value=None), \
mock.patch.object(sp, "push_env", lambda *a: calls.append(a) or 0), \
mock.patch("sys.stdin") as stdin, \
contextlib.redirect_stderr(io.StringIO()) as err:
stdin.isatty.return_value = tty
code = sp.push_env_command(CFG, args)
return code, calls, err.getvalue()
def test_remote_target_with_options(self):
code, calls, _ = self.run_cmd(["1", "igor@h:/p/.env", "--port", "2222", "--key", "/k", "--dry-run"])
self.assertEqual(code, 0)
_, n, spec, _, dry = calls[0]
self.assertEqual((n, dry), (1, True))
self.assertEqual(spec, {"kind": "ssh", "user": "igor", "host": "h", "path": "/p/.env", "port": 2222, "key": "/k"})
def test_local_target(self):
_, calls, _ = self.run_cmd(["2", "/srv/app/.env"])
self.assertEqual(calls[0][2], {"kind": "local", "path": "/srv/app/.env"})
def test_unknown_instance(self):
code, calls, err = self.run_cmd(["3", "/x"])
self.assertEqual((code, calls), (1, []))
self.assertIn("Instance 3 does not exist", err)
def test_missing_target_without_a_terminal(self):
code, calls, err = self.run_cmd(["1"])
self.assertEqual((code, calls), (1, []))
self.assertIn("Say where the file is", err)
def test_windows_path_without_computer(self):
code, calls, err = self.run_cmd(["1", "C:\\Users\\x\\.env"])
self.assertEqual((code, calls), (1, []))
self.assertIn("Windows path without the computer's name", err)
if __name__ == "__main__":
unittest.main()