#!/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())