diff --git a/manage.py b/manage.py new file mode 100644 index 0000000..99b37f0 --- /dev/null +++ b/manage.py @@ -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())