108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
"""CLI для управления таблицей токенов OpenCloud заказчиков.
|
||
|
||
Использование:
|
||
python scripts/manage_opencloud_tokens.py add <username>
|
||
— интерактивно спросит App Token (ввод скрыт, getpass).
|
||
python scripts/manage_opencloud_tokens.py list
|
||
— список username, для которых сохранён токен.
|
||
python scripts/manage_opencloud_tokens.py remove <username>
|
||
— удаляет запись (например, при увольнении/отзыве доступа).
|
||
|
||
Требует настроенные OPENBAO__SETTINGS__DATABASE_URL и
|
||
OPENBAO__SETTINGS__FERNET_KEY (или .env для локального теста).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import getpass
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
from app.opencloud_token_store import ( # noqa: E402
|
||
OpenCloudTokenStoreError,
|
||
add_token,
|
||
get_token,
|
||
list_usernames,
|
||
remove_token,
|
||
)
|
||
|
||
|
||
async def _cmd_add(username: str) -> None:
|
||
token = getpass.getpass(f"App Token для '{username}' (ввод скрыт): ").strip()
|
||
if not token:
|
||
print("Пустой токен — отменено.")
|
||
return
|
||
added_by = getpass.getuser()
|
||
await add_token(username, token, added_by=added_by)
|
||
print(f"Сохранено: {username}")
|
||
|
||
|
||
async def _cmd_list() -> None:
|
||
usernames = await list_usernames()
|
||
if not usernames:
|
||
print("Таблица пуста.")
|
||
return
|
||
for username in usernames:
|
||
print(username)
|
||
|
||
|
||
async def _cmd_remove(username: str) -> None:
|
||
removed = await remove_token(username)
|
||
print("Удалено." if removed else "Запись не найдена.")
|
||
|
||
|
||
async def _cmd_check(username: str) -> None:
|
||
"""Проверка без вывода самого токена — только факт наличия и что он расшифровывается."""
|
||
token = await get_token(username)
|
||
if token is None:
|
||
print(f"Токена для '{username}' нет в таблице.")
|
||
else:
|
||
print(
|
||
f"Токен для '{username}' найден и успешно расшифрован ({len(token)} симв.)."
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
add_parser = subparsers.add_parser("add", help="Добавить/обновить токен заказчика")
|
||
add_parser.add_argument("username")
|
||
|
||
subparsers.add_parser("list", help="Список username с сохранённым токеном")
|
||
|
||
remove_parser = subparsers.add_parser("remove", help="Удалить токен заказчика")
|
||
remove_parser.add_argument("username")
|
||
|
||
check_parser = subparsers.add_parser(
|
||
"check", help="Проверить наличие и расшифровываемость токена"
|
||
)
|
||
check_parser.add_argument("username")
|
||
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
if args.command == "add":
|
||
asyncio.run(_cmd_add(args.username))
|
||
elif args.command == "list":
|
||
asyncio.run(_cmd_list())
|
||
elif args.command == "remove":
|
||
asyncio.run(_cmd_remove(args.username))
|
||
elif args.command == "check":
|
||
asyncio.run(_cmd_check(args.username))
|
||
except OpenCloudTokenStoreError as exc:
|
||
print(f"Ошибка: {exc}", file=sys.stderr)
|
||
raise SystemExit(1) from exc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|