server-side/app/cli.py - master - internetfriend/seif

server-side/app/cli.py
196 строк8 KB
internetfriend
seif initial preview
26 июл 2026, 15:23
26 июл 2026, 15:231bcc0dd
"""Local administration for СЕЙФ.

Creating the first user over HTTP needs the X-Admin-Token header, which is
awkward from Swagger and impossible from a plain browser. This does the same
work directly against the database, on the machine that already has the
credentials — the usual way to bootstrap a service of this kind.

    python -m app.cli create-user --login a.petrov \\
        --full-name "Петров Андрей Сергеевич" --secret 'vault-secret-1' \\
        --email a.petrov@example.ru --department "УИБ" --admin

    python -m app.cli list-users
    python -m app.cli set-secret --login a.petrov --secret 'new-secret'
    python -m app.cli admin-token

The service does not need to be running. SQLite in WAL mode tolerates it
running at the same time.
"""

from __future__ import annotations

import argparse
import sys

from .config import get_settings
from .db import connect, init_db
from .models import User, new_id, now, now_iso
from .routers.admin import derive_names
from .security import hash_secret


def create_user(args: argparse.Namespace) -> int:
    settings = get_settings()

    if args.secret and len(args.secret) < settings.min_secret_length:
        print(
            f"Секретная строка должна быть не короче {settings.min_secret_length} символов.",
            file=sys.stderr,
        )
        return 1

    db = connect()
    try:
        if db.execute("SELECT 1 FROM users WHERE login = ?", (args.login,)).fetchone():
            print(f"Учётная запись «{args.login}» уже существует.", file=sys.stderr)
            return 1

        short_name, initials = derive_names(args.full_name, args.short_name, args.initials)
        user_id = new_id("usr")
        db.execute(
            "INSERT INTO users (id, login, full_name, short_name, initials, email,"
            " department, position, secret_hash, is_admin, created_at)"
            " VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            (user_id, args.login, args.full_name, short_name, initials, args.email,
             args.department, args.position,
             hash_secret(args.secret) if args.secret else None,
             int(args.admin), now_iso()),
        )
        db.execute(
            "INSERT INTO audit_events (id, at, actor_id, action, target, namespace,"
            " result, address) VALUES (?,?,NULL,?,?,?,?,?)",
            (new_id("evt"), now_iso(), "Создана учётная запись", args.login,
             "personal", "ok", "cli"),
        )
        db.commit()
    finally:
        db.close()

    print(f"Создана учётная запись {args.login} ({short_name}), id {user_id}")
    if not args.secret:
        print("Секретная строка не задана: вход возможен только по сертификату.")
    return 0


def list_users(_: argparse.Namespace) -> int:
    db = connect()
    try:
        rows = db.execute("SELECT * FROM users ORDER BY created_at").fetchall()
    finally:
        db.close()

    if not rows:
        print("Пользователей нет. Создайте первого: python -m app.cli create-user --help")
        return 0

    users = [User.from_row(row) for row in rows]
    width = max(len(user.login) for user in users)
    moment = now()
    for user in users:
        flags = []
        if user.is_admin:
            flags.append("админ")
        if not user.is_active:
            flags.append("отключён")
        if not user.secret_hash:
            flags.append("только сертификат")
        if user.locked_until and user.locked_until > moment:
            flags.append("заблокирован")
        suffix = f"  [{', '.join(flags)}]" if flags else ""
        print(f"{user.login:<{width}}  {user.full_name}{suffix}")
    return 0


def set_secret(args: argparse.Namespace) -> int:
    settings = get_settings()
    if len(args.secret) < settings.min_secret_length:
        print(
            f"Секретная строка должна быть не короче {settings.min_secret_length} символов.",
            file=sys.stderr,
        )
        return 1

    db = connect()
    try:
        row = db.execute("SELECT id FROM users WHERE login = ?", (args.login,)).fetchone()
        if row is None:
            print(f"Учётная запись «{args.login}» не найдена.", file=sys.stderr)
            return 1

        user_id = row["id"]
        # A secret reset also clears a lockout and kills live sessions.
        db.execute(
            "UPDATE users SET secret_hash = ?, failed_attempts = 0, locked_until = NULL"
            " WHERE id = ?",
            (hash_secret(args.secret), user_id),
        )
        cursor = db.execute(
            "UPDATE sessions SET revoked_at = ? WHERE user_id = ? AND revoked_at IS NULL",
            (now_iso(), user_id),
        )
        revoked = cursor.rowcount
        db.commit()
    finally:
        db.close()

    print(f"Секретная строка изменена для {args.login}. Завершено сессий: {revoked}.")
    return 0


def admin_token(_: argparse.Namespace) -> int:
    settings = get_settings()
    if settings.admin_token:
        print(settings.admin_token)
        return 0

    print(
        "SEIF_ADMIN_TOKEN не задан. Сервис генерирует его при каждом запуске и печатает\n"
        "в stderr, поэтому здесь его получить нельзя — значение принадлежит тому процессу.\n"
        "Задайте постоянное значение или пользуйтесь этой утилитой вместо HTTP.",
        file=sys.stderr,
    )
    return 1


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="python -m app.cli",
        description="Локальное администрирование СЕЙФ (без HTTP и заголовков).",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    create = sub.add_parser("create-user", help="создать учётную запись")
    create.add_argument("--login", required=True)
    create.add_argument("--full-name", required=True, help="Фамилия Имя Отчество")
    create.add_argument("--secret", default="", help="секретная строка; без неё — только сертификат")
    create.add_argument("--email", default="")
    create.add_argument("--department", default="")
    create.add_argument("--position", default="")
    create.add_argument("--short-name", default="", help="по умолчанию выводится из ФИО")
    create.add_argument("--initials", default="", help="по умолчанию выводятся из ФИО")
    create.add_argument("--admin", action="store_true", help="признак администратора")
    create.set_defaults(handler=create_user)

    listing = sub.add_parser("list-users", help="показать учётные записи")
    listing.set_defaults(handler=list_users)

    secret = sub.add_parser("set-secret", help="задать секретную строку")
    secret.add_argument("--login", required=True)
    secret.add_argument("--secret", required=True)
    secret.set_defaults(handler=set_secret)

    token = sub.add_parser("admin-token", help="показать административный токен")
    token.set_defaults(handler=admin_token)

    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    # Creates the schema, so the CLI works on a fresh checkout.
    init_db()
    return args.handler(args)


if __name__ == "__main__":
    raise SystemExit(main())