452 lines
15 KiB
Python
452 lines
15 KiB
Python
"""CLI entry point for volumen."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import getpass
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
import warnings
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from . import VERSION
|
|
|
|
PYPI_URL = "https://pypi.org/pypi/volumen/json"
|
|
PYPI_TIMEOUT = 5.0
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
prog="volumen", description="A small, file-based Markdown blog engine."
|
|
)
|
|
parser.add_argument("--version", action="version", version=f"volumen {VERSION}")
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
|
serve_parser = sub.add_parser("serve", help="Start the blog server")
|
|
serve_parser.add_argument("--config", default="/etc/volumen/config.toml")
|
|
serve_parser.add_argument("--host", default=None)
|
|
serve_parser.add_argument("--port", type=int, default=None)
|
|
serve_parser.add_argument("--content", default=None)
|
|
|
|
hash_parser = sub.add_parser("hash-password", help="Hash a password")
|
|
hash_parser.add_argument(
|
|
"password",
|
|
nargs="?",
|
|
default=None,
|
|
help=(
|
|
"Optional plaintext password (discouraged — visible in shell "
|
|
"history). If omitted you will be prompted."
|
|
),
|
|
)
|
|
|
|
init_parser = sub.add_parser(
|
|
"init",
|
|
help=(
|
|
"Bootstrap config, data directories, admin password (and optionally a systemd service)"
|
|
),
|
|
)
|
|
init_parser.add_argument(
|
|
"--systemd",
|
|
action="store_true",
|
|
help="Install / enable a systemd unit (root only)",
|
|
)
|
|
init_parser.add_argument(
|
|
"--local",
|
|
action="store_true",
|
|
help="Use per-user paths (~/.config/volumen/, ~/.local/share/volumen/)",
|
|
)
|
|
init_parser.add_argument(
|
|
"--config",
|
|
type=Path,
|
|
default=None,
|
|
help=(
|
|
"Override config path (default: /etc/volumen/config.toml when "
|
|
"system, ~/.config/volumen/config.toml when user)"
|
|
),
|
|
)
|
|
init_parser.add_argument("--data", type=Path, default=None, help="Override data directory")
|
|
init_parser.add_argument(
|
|
"--users-file", type=Path, default=None, help="Override users.toml path"
|
|
)
|
|
init_parser.add_argument(
|
|
"--service-user", default="volumen", help="System user for systemd unit (default: volumen)"
|
|
)
|
|
enable_group = init_parser.add_mutually_exclusive_group()
|
|
enable_group.add_argument(
|
|
"--enable",
|
|
action="store_true",
|
|
dest="enable",
|
|
help="With --systemd, also systemctl enable --now (default when root)",
|
|
)
|
|
enable_group.add_argument(
|
|
"--no-enable",
|
|
action="store_false",
|
|
dest="enable",
|
|
help="With --systemd, install but do not enable",
|
|
)
|
|
init_parser.set_defaults(enable=True)
|
|
init_parser.add_argument(
|
|
"--admin-password",
|
|
default=None,
|
|
help="Set admin password non-interactively (discouraged)",
|
|
)
|
|
init_parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Overwrite existing config (with .bak.<ts> backup)",
|
|
)
|
|
|
|
status_parser = sub.add_parser("status", help="Show installation status")
|
|
status_parser.add_argument("--config", type=Path, default=None, help="Config path to inspect")
|
|
status_parser.add_argument(
|
|
"--data", type=Path, default=None, help="Data dir (posts) to inspect"
|
|
)
|
|
status_parser.add_argument(
|
|
"--users-file", type=Path, default=None, help="users.toml path to inspect"
|
|
)
|
|
status_parser.add_argument("--json", action="store_true", help="Machine-readable output")
|
|
|
|
check_update_parser = sub.add_parser(
|
|
"check-update",
|
|
help="Compare the installed version with the latest release on PyPI",
|
|
)
|
|
check_update_parser.add_argument(
|
|
"--json", action="store_true", help="Machine-readable output (sets exit code only)"
|
|
)
|
|
|
|
sub.add_parser("version", help="Show version")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "version":
|
|
print(VERSION)
|
|
elif args.command == "hash-password":
|
|
_hash_password(args)
|
|
elif args.command == "init":
|
|
_init(args)
|
|
elif args.command == "status":
|
|
sys.exit(_status(args))
|
|
elif args.command == "check-update":
|
|
sys.exit(_check_update(args))
|
|
elif args.command == "serve":
|
|
_serve(args)
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
|
|
def _hash_password(args: argparse.Namespace) -> None:
|
|
from .password import hash_password
|
|
|
|
if args.password is not None:
|
|
warnings.warn(
|
|
"passing the password as a positional argument is insecure "
|
|
"(it ends up in shell history). Prefer interactive input.",
|
|
stacklevel=2,
|
|
)
|
|
pw = args.password
|
|
else:
|
|
pw = getpass.getpass("Password: ")
|
|
print(hash_password(pw))
|
|
|
|
|
|
def _init(args: argparse.Namespace) -> None:
|
|
from .installer import (
|
|
SYSTEM_CONFIG_PATH,
|
|
SYSTEM_CONTENT_DIR,
|
|
SYSTEM_USERS_FILE,
|
|
USER_CONFIG_PATH,
|
|
USER_CONTENT_DIR,
|
|
USER_USERS_FILE,
|
|
InstallerError,
|
|
system_install,
|
|
user_install,
|
|
)
|
|
|
|
is_root = _is_root()
|
|
use_user = bool(args.local) or not is_root
|
|
install_service = bool(args.systemd) and not use_user
|
|
|
|
if use_user and args.systemd:
|
|
print(
|
|
"error: --systemd is not available for per-user installs; "
|
|
"drop --local and run as root.",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(2)
|
|
|
|
if use_user:
|
|
config_path = args.config if args.config is not None else USER_CONFIG_PATH
|
|
content_dir = args.data if args.data is not None else USER_CONTENT_DIR
|
|
users_file = args.users_file if args.users_file is not None else USER_USERS_FILE
|
|
if args.data is None:
|
|
# Per-user defaults: ~/.local/share/volumen/{posts,users.toml}.
|
|
content_dir = USER_CONTENT_DIR
|
|
users_file = USER_USERS_FILE
|
|
else:
|
|
config_path = args.config if args.config is not None else SYSTEM_CONFIG_PATH
|
|
content_dir = args.data if args.data is not None else SYSTEM_CONTENT_DIR
|
|
users_file = args.users_file if args.users_file is not None else SYSTEM_USERS_FILE
|
|
|
|
password = args.admin_password
|
|
if password is None:
|
|
while True:
|
|
first = getpass.getpass("Admin password (user 'admin'): ")
|
|
second = getpass.getpass("Confirm: ")
|
|
if first != second:
|
|
print("Passwords do not match; try again.", file=sys.stderr)
|
|
continue
|
|
if not first:
|
|
print("Password cannot be empty.", file=sys.stderr)
|
|
continue
|
|
password = first
|
|
break
|
|
|
|
common: dict[str, object] = {
|
|
"config_path": Path(config_path),
|
|
"content_dir": Path(content_dir),
|
|
"users_file": Path(users_file),
|
|
"admin_password": password,
|
|
"force": bool(args.force),
|
|
}
|
|
|
|
try:
|
|
if use_user:
|
|
result = user_install(**common)
|
|
else:
|
|
result = system_install(
|
|
**common,
|
|
service_user=args.service_user,
|
|
install_service=install_service,
|
|
enable_service=bool(args.enable),
|
|
)
|
|
except InstallerError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
_print_init_summary(result, install_service=install_service, use_user=use_user)
|
|
|
|
|
|
def _print_init_summary(result: object, *, install_service: bool, use_user: bool) -> None:
|
|
from .installer import InstallResult
|
|
|
|
assert isinstance(result, InstallResult)
|
|
skipped = any("leaving untouched" in m for m in result.messages)
|
|
if skipped:
|
|
print("==> volumen: existing config left untouched (use --force to overwrite).")
|
|
print()
|
|
print(f" Config: {result.config_path}")
|
|
print()
|
|
print("Nothing to do. Re-run with --force if you want to re-bootstrap.")
|
|
return
|
|
|
|
print("==> volumen installed.")
|
|
print()
|
|
print(f" Config: {result.config_path}")
|
|
if result.service_user is not None:
|
|
print(f" (chmod 600, owned by {result.service_user})")
|
|
print(f" Data dir: {result.content_dir}")
|
|
if result.service_user is not None:
|
|
print(f" (owned by {result.service_user})")
|
|
print(f" Users file: {result.users_file}")
|
|
if result.service_user is not None:
|
|
print(f" (owned by {result.service_user})")
|
|
if install_service and result.systemd_unit_path is not None:
|
|
print(f" Service: systemd unit installed at {result.systemd_unit_path}")
|
|
print(" (enabled, running)")
|
|
else:
|
|
print(" Service: not installed")
|
|
print(" Admin user: admin (password you just typed)")
|
|
print()
|
|
print("Next steps:")
|
|
print(f" 1. Edit {result.config_path} — set site.base_url, site.title, etc.")
|
|
print(" 2. Set up nginx as reverse proxy with TLS (see docs/deployment.md).")
|
|
if install_service and result.systemd_unit_path is not None:
|
|
print(" 3. systemctl status volumen")
|
|
|
|
|
|
def _status(args: argparse.Namespace) -> int:
|
|
from .installer import (
|
|
SYSTEM_CONFIG_PATH,
|
|
SYSTEMD_UNIT_PATH,
|
|
inspect,
|
|
)
|
|
|
|
config_path = Path(args.config) if args.config is not None else SYSTEM_CONFIG_PATH
|
|
content_dir = Path(args.data) if args.data is not None else _content_dir_from(config_path)
|
|
users_file = (
|
|
Path(args.users_file) if args.users_file is not None else _users_file_from(config_path)
|
|
)
|
|
info = inspect(
|
|
config_path=config_path,
|
|
content_dir=content_dir,
|
|
users_file=users_file,
|
|
systemd_unit_path=SYSTEMD_UNIT_PATH,
|
|
)
|
|
|
|
if args.json:
|
|
print(json.dumps(info, indent=2, sort_keys=True))
|
|
return 0 if not info["issues"] else 1
|
|
|
|
print("==> volumen status")
|
|
print()
|
|
print(f" Config: {info['config_path']}")
|
|
print(f" Data dir: {info['data_dir']}")
|
|
print(f" Users file: {info['users_file']}")
|
|
print(
|
|
f" Systemd unit: {'installed' if info['systemd_unit_installed'] else 'not installed'}"
|
|
)
|
|
print(f" Service: {info['service_active']}")
|
|
print(f" Admin user: {'present' if info['admin_user_present'] else 'missing'}")
|
|
print()
|
|
issues = list(info["issues"])
|
|
if not issues:
|
|
print("All checks passed.")
|
|
else:
|
|
print("Issues:")
|
|
for issue in issues:
|
|
print(f" - {issue}")
|
|
return 0 if not issues else 1
|
|
|
|
|
|
def _serve(args: argparse.Namespace) -> None:
|
|
from .app import create_app
|
|
from .config import Config
|
|
from .store import Store
|
|
|
|
overrides: dict[str, object] = {}
|
|
if args.host:
|
|
overrides["host"] = args.host
|
|
if args.port:
|
|
overrides["port"] = args.port
|
|
if args.content:
|
|
overrides["content"] = args.content
|
|
|
|
config = Config.load(path=args.config, overrides=overrides)
|
|
store = Store(config.content_dir, default_lang=config.language)
|
|
app = create_app(config, store)
|
|
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host=config.host, port=config.port, log_level="info")
|
|
|
|
|
|
def _content_dir_from(config_path: Path) -> Path:
|
|
"""Best-effort resolution of the ``content_dir`` for a given config file.
|
|
|
|
Reads the value from the loaded config if the file is present and
|
|
parseable; otherwise returns the conventional system default. Used by
|
|
``volumen status`` so the operator doesn't have to pass ``--data``
|
|
separately.
|
|
"""
|
|
from .config import Config
|
|
from .installer import SYSTEM_CONTENT_DIR
|
|
|
|
try:
|
|
return Path(Config.load(path=str(config_path)).content_dir)
|
|
except OSError, ValueError:
|
|
return SYSTEM_CONTENT_DIR
|
|
|
|
|
|
def _users_file_from(config_path: Path) -> Path:
|
|
from .config import Config
|
|
from .installer import SYSTEM_USERS_FILE
|
|
|
|
try:
|
|
return Path(Config.load(path=str(config_path)).users_file)
|
|
except OSError, ValueError:
|
|
return SYSTEM_USERS_FILE
|
|
|
|
|
|
def _is_root() -> bool:
|
|
try:
|
|
return os.geteuid() == 0
|
|
except AttributeError:
|
|
return False
|
|
|
|
|
|
def _get_installed_version() -> str:
|
|
"""Return the installed `volumen` distribution version, or a sentinel."""
|
|
from importlib.metadata import PackageNotFoundError
|
|
from importlib.metadata import version as _pkg_version
|
|
|
|
try:
|
|
return _pkg_version("volumen")
|
|
except PackageNotFoundError: # pragma: no cover — only reached when not installed as a wheel.
|
|
return "0.0.0+unknown"
|
|
|
|
|
|
def _compare_versions(installed: str, latest: str) -> bool:
|
|
"""Return True if `latest` is strictly newer than `installed`.
|
|
|
|
Pre-release / local segments (``0.5.0a1``, ``0.5.0.dev1``, ``1.0.0+abc``) are
|
|
stripped before comparison so a published ``1.0.0a1`` is correctly reported
|
|
as an update against an installed ``0.9.0``.
|
|
"""
|
|
|
|
def _core(v: str) -> tuple[int, ...]:
|
|
for sep in ("a", "b", "rc", ".dev", "+"):
|
|
v = v.split(sep, 1)[0]
|
|
try:
|
|
return tuple(int(p) for p in v.split("."))
|
|
except ValueError:
|
|
return (0,)
|
|
|
|
return _core(latest) > _core(installed)
|
|
|
|
|
|
def _check_update(args: argparse.Namespace) -> int:
|
|
"""Check if a newer `volumen` is available on PyPI.
|
|
|
|
Exit codes: ``0`` up to date; ``1`` update available; ``2`` PyPI unreachable
|
|
(offline, timeout, malformed response, or local-only install).
|
|
"""
|
|
current = _get_installed_version()
|
|
checked_at = datetime.now(UTC).isoformat()
|
|
|
|
latest: str | None = None
|
|
error: str | None = None
|
|
try:
|
|
with urllib.request.urlopen(PYPI_URL, timeout=PYPI_TIMEOUT) as resp:
|
|
payload = json.loads(resp.read())
|
|
latest = payload["info"]["version"]
|
|
except (
|
|
urllib.error.URLError,
|
|
urllib.error.HTTPError,
|
|
OSError,
|
|
json.JSONDecodeError,
|
|
KeyError,
|
|
) as exc:
|
|
error = str(exc) or exc.__class__.__name__
|
|
|
|
update_available = latest is not None and _compare_versions(current, latest)
|
|
|
|
if args.json:
|
|
out: dict[str, object] = {
|
|
"current_version": current,
|
|
"latest_version": latest,
|
|
"update_available": update_available,
|
|
"checked_at": checked_at,
|
|
}
|
|
if error is not None:
|
|
out["error"] = error
|
|
print(json.dumps(out, indent=2, sort_keys=True))
|
|
return 1 if update_available else (2 if latest is None else 0)
|
|
|
|
if latest is None:
|
|
print(
|
|
f"volumen {current} — could not reach {PYPI_URL} ({error})",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
if update_available:
|
|
print(f"volumen {current} (latest: {latest}) — update available")
|
|
print(" Run: uv tool upgrade volumen")
|
|
return 1
|
|
print(f"volumen {current} (up to date)")
|
|
return 0
|