feat(cli): add setup, checks and portable backups
Add interactive administrator setup, installation checks and consistent SQLite/media backups.
This commit is contained in:
rekushi committed
1 parent
7440a1061c
commit
cad8c3650a
1 file changed
+175
@@ -0,0 +1,175 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Set up, check and back up a RexiMemo OSE installation."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from contextlib import closing
|
||||
from pathlib import Path
|
||||
import argparse
|
||||
import getpass
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import socket
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import zipfile
|
||||
|
||||
from server_config import ROOT, config_path, load_config, save_config
|
||||
|
||||
|
||||
def database_path() -> Path:
|
||||
configured = Path(os.environ.get("REXIMEMO_DB", ROOT / "database" / "reximemo.sqlite3"))
|
||||
return configured.resolve() if configured.is_absolute() else (ROOT / configured).resolve()
|
||||
|
||||
|
||||
def setup_server(*, input_fn=input, password_fn=getpass.getpass) -> None:
|
||||
settings = load_config()
|
||||
print("\nSet up your Flipnote server\n")
|
||||
print("This keeps the bundled sample Flipnotes and existing accounts.")
|
||||
print("You can change branding, channels and posting options later in Admin.\n")
|
||||
name = input_fn(f"Server name [{settings['site_name']}]: ").strip() or settings["site_name"]
|
||||
port = input_fn(f"Port [{settings['port']}]: ").strip() or settings["port"]
|
||||
host = input_fn(f"Listen address [{settings['host']}]: ").strip() or settings["host"]
|
||||
username = input_fn("Administrator username [admin]: ").strip() or "admin"
|
||||
password = password_fn("Administrator password (at least 8 characters): ")
|
||||
confirm = password_fn("Enter the password again: ")
|
||||
if password != confirm:
|
||||
raise ValueError("The passwords did not match. Run setup again.")
|
||||
if len(password) < 8:
|
||||
raise ValueError("Use an administrator password of at least 8 characters.")
|
||||
from server_config import validate_config
|
||||
validate_config({**settings, "site_name": name, "host": host, "port": port})
|
||||
from database import Database
|
||||
account = Database.get_account_by_username(username)
|
||||
if account:
|
||||
if username.lower() != "reximemo" and not Database.verify_account(username, password):
|
||||
raise ValueError("That account already exists. Enter its current password or choose another username.")
|
||||
account_id = int(account["id"])
|
||||
else:
|
||||
account_id = Database.create_account(username, password)
|
||||
disabled_default = False
|
||||
with Database.lock, Database.conn:
|
||||
Database.conn.execute("UPDATE accounts SET is_admin=1,password_sha256=? WHERE id=?",
|
||||
(Database._password_hash(password), account_id))
|
||||
default = Database.get_account_by_username("reximemo")
|
||||
if (default and int(default["id"]) != account_id
|
||||
and default["password_sha256"] == hashlib.sha256(b"alpine").hexdigest()):
|
||||
Database.conn.execute("UPDATE accounts SET is_admin=0,password_sha256=? WHERE id=?",
|
||||
(Database._password_hash(secrets.token_urlsafe(48)), default["id"]))
|
||||
Database.conn.execute("DELETE FROM ip_logins WHERE account_id=?", (default["id"],))
|
||||
disabled_default = True
|
||||
save_config({"site_name": name, "port": port, "host": host, "setup_complete": True})
|
||||
print(f"\nSetup complete. Sign in as {username} at http://127.0.0.1:{int(port)}/login")
|
||||
if disabled_default:
|
||||
print("The bundled reximemo/alpine login has been disabled.")
|
||||
print("Run Start-OSE.ps1 on Windows or ./start-ose.sh on Linux/macOS to start the server.")
|
||||
|
||||
|
||||
def check_installation() -> bool:
|
||||
print("RexiMemo OSE installation check\n")
|
||||
okay = sys.version_info >= (3, 10)
|
||||
print(("OK" if okay else "FIX") + f" Python {sys.version_info.major}.{sys.version_info.minor} (3.10 or newer required)")
|
||||
for module, package in (("twisted", "Twisted"), ("numpy", "NumPy"), ("PIL", "Pillow")):
|
||||
present = importlib.util.find_spec(module) is not None
|
||||
print(("OK" if present else "FIX") + " " + package + (" installed" if present else " missing; run the start script to install dependencies"))
|
||||
okay = okay and present
|
||||
settings = load_config()
|
||||
print(f"OK Settings: {config_path()}" if config_path().exists() else "NOTE First-run setup has not been saved yet.")
|
||||
path = database_path()
|
||||
if path.is_file():
|
||||
with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)) as database:
|
||||
valid = database.execute("PRAGMA quick_check").fetchone()[0] == "ok"
|
||||
print(("OK" if valid else "FIX") + " SQLite database")
|
||||
okay = okay and valid
|
||||
else:
|
||||
print("FIX Database is missing: " + str(path))
|
||||
okay = False
|
||||
host = os.environ.get("REXIMEMO_HOST", settings["host"])
|
||||
port = int(os.environ.get("REXIMEMO_PORT", settings["port"]))
|
||||
family = socket.AF_INET6 if ":" in host else socket.AF_INET
|
||||
with socket.socket(family, socket.SOCK_STREAM) as probe:
|
||||
try:
|
||||
probe.bind((host, port))
|
||||
print(f"OK Listen address {host}:{port} is available")
|
||||
except OSError:
|
||||
print(f"NOTE {host}:{port} is in use. If OSE is already running, this is expected.")
|
||||
with tempfile.TemporaryFile(dir=ROOT):
|
||||
pass
|
||||
print("OK Installation folder is writable")
|
||||
if not settings["setup_complete"]:
|
||||
print("NEXT Run: python manage.py setup")
|
||||
print("\nDSi: use a working DNS/NAS service, then point the connection's HTTP proxy at this computer and the port shown above.")
|
||||
return okay
|
||||
|
||||
|
||||
def backup_server(destination: Path | None = None) -> Path:
|
||||
"""Make a portable SQLite snapshot plus its media and branding files."""
|
||||
source = database_path()
|
||||
if not source.is_file():
|
||||
raise ValueError("The database was not found. Check REXIMEMO_DB before backing up.")
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||||
destination = Path(destination) if destination else ROOT / "backups" / f"reximemo-ose-{stamp}.zip"
|
||||
if not destination.is_absolute():
|
||||
destination = ROOT / destination
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
if destination.exists():
|
||||
raise ValueError("That backup already exists. Choose a different filename.")
|
||||
with tempfile.TemporaryDirectory(prefix="ose-backup-") as directory:
|
||||
snapshot = Path(directory) / "reximemo.sqlite3"
|
||||
with closing(sqlite3.connect(source.as_uri() + "?mode=ro", uri=True)) as database, closing(sqlite3.connect(snapshot)) as saved:
|
||||
with saved:
|
||||
database.backup(saved)
|
||||
files = {"database/reximemo.sqlite3": snapshot}
|
||||
for relative in ("database/Creators", "database/Comments", "web/static", "hatenadir/images", "hatenadir/css"):
|
||||
folder = ROOT / relative
|
||||
if folder.is_dir():
|
||||
for path in folder.rglob("*"):
|
||||
if path.is_file() and not path.is_symlink():
|
||||
files[path.relative_to(ROOT).as_posix()] = path
|
||||
if config_path().is_file():
|
||||
files["server-settings.json"] = config_path()
|
||||
logo = source.parent / "site-logo.png"
|
||||
if logo.is_file():
|
||||
files["database/site-logo.png"] = logo
|
||||
try:
|
||||
with zipfile.ZipFile(destination, "x", compression=zipfile.ZIP_DEFLATED) as archive:
|
||||
for relative, path in sorted(files.items()):
|
||||
archive.write(path, relative)
|
||||
archive.writestr("BACKUP-README.txt", "RexiMemo OSE backup\n\nStop the server before restoring. Keep a copy of the current installation, then extract these files into the OSE folder. This archive contains accounts, IP login records and private data: keep it private. If you use REXIMEMO_DB or REXIMEMO_CONFIG, restore the corresponding files to those configured locations.\n")
|
||||
destination.chmod(0o600)
|
||||
except Exception:
|
||||
destination.unlink(missing_ok=True)
|
||||
raise
|
||||
print("Backup saved: " + str(destination))
|
||||
return destination
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("command", nargs="?", choices=("setup", "check", "backup"))
|
||||
parser.add_argument("--check", action="store_true", help="check dependencies, configuration and the database")
|
||||
parser.add_argument("--backup", action="store_true", help="make a backup")
|
||||
parser.add_argument("--output", type=Path, help="backup ZIP filename")
|
||||
arguments = parser.parse_args(argv)
|
||||
command = "check" if arguments.check else "backup" if arguments.backup else arguments.command
|
||||
try:
|
||||
if command == "setup":
|
||||
setup_server()
|
||||
elif command == "check":
|
||||
return 0 if check_installation() else 1
|
||||
elif command == "backup":
|
||||
backup_server(arguments.output)
|
||||
else:
|
||||
parser.print_help()
|
||||
return 0
|
||||
except (OSError, ValueError, sqlite3.Error) as exc:
|
||||
print("\nCould not complete that: " + str(exc), file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user