patrickandClaude Sonnet 5.5
f34fc6d686
- LoggingAdapter protokolliert send/edit (Matrix zusätzlich reply) inkl. Fehlversuche - Tabelle outbound_log (Migration 10), Aufbewahrung retention.outbound_days (30 Tage) - Seite /messages und CLI checkbot messages mit Filtern - Text gekappt (4000) und Secrets maskiert Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
import argparse
|
|
import getpass
|
|
import os
|
|
import secrets
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from .crypto import SecretBox, generate_master_key
|
|
from .db import Database
|
|
from .fileconfig import MASTER_KEY_FILE, SESSION_SECRET_FILE, load_file_config, load_master_key
|
|
from .outbound import OutboundRepository, parse_since
|
|
from .users import ROLE_ADMIN, UserError, UserStore
|
|
|
|
CLI_ACTOR = "cli"
|
|
SECRET_FILE_MODE = 0o600
|
|
CONFIG_DIR_MODE = 0o700
|
|
SESSION_SECRET_BYTES = 48
|
|
MESSAGES_TEXT_PREVIEW = 200
|
|
|
|
|
|
def _write_secret_file(path: Path, content: str) -> bool:
|
|
if path.exists():
|
|
return False
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=CONFIG_DIR_MODE)
|
|
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, SECRET_FILE_MODE)
|
|
with os.fdopen(descriptor, "w") as handle:
|
|
handle.write(content)
|
|
return True
|
|
|
|
|
|
def _user_store() -> UserStore:
|
|
config = load_file_config()
|
|
box = SecretBox(load_master_key(config.config_dir / MASTER_KEY_FILE))
|
|
return UserStore(Database(config.db_path), box)
|
|
|
|
|
|
def _ask_password() -> str:
|
|
password = getpass.getpass("Passwort: ")
|
|
if password != getpass.getpass("Passwort wiederholen: "):
|
|
raise UserError("Passwörter stimmen nicht überein")
|
|
return password
|
|
|
|
|
|
def cmd_gen_keys(args: argparse.Namespace) -> int:
|
|
config = load_file_config()
|
|
files = {
|
|
MASTER_KEY_FILE: generate_master_key(),
|
|
SESSION_SECRET_FILE: secrets.token_urlsafe(SESSION_SECRET_BYTES),
|
|
}
|
|
for name, content in files.items():
|
|
created = _write_secret_file(config.config_dir / name, content)
|
|
print(f"{name}: {'angelegt' if created else 'existiert bereits, unverändert'}")
|
|
return 0
|
|
|
|
|
|
def cmd_create_admin(args: argparse.Namespace) -> int:
|
|
username = args.username or input("Benutzername: ").strip()
|
|
try:
|
|
_user_store().create(username, _ask_password(), ROLE_ADMIN, actor=CLI_ACTOR)
|
|
except UserError as exc:
|
|
print(f"Fehler: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(f"Admin '{username}' angelegt")
|
|
return 0
|
|
|
|
|
|
def cmd_messages(args: argparse.Namespace) -> int:
|
|
config = load_file_config()
|
|
repo = OutboundRepository(Database(config.db_path))
|
|
since = parse_since(args.since) if args.since else None
|
|
try:
|
|
entries = repo.search(backend=args.backend, query=args.grep, only_errors=args.errors, since=since,
|
|
limit=args.limit)
|
|
except ValueError as exc:
|
|
print(f"Fehler: {exc}", file=sys.stderr)
|
|
return 1
|
|
for entry in entries:
|
|
iso = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(entry.ts))
|
|
status = "ok" if entry.ok else f"ERR:{entry.error or ''}"
|
|
text = entry.text[:MESSAGES_TEXT_PREVIEW].replace("\n", "\\n")
|
|
print(f"{iso} {entry.backend} {entry.target} {entry.action} {status} {text}")
|
|
return 0
|
|
|
|
|
|
def cmd_set_password(args: argparse.Namespace) -> int:
|
|
users = _user_store()
|
|
user = users.get_by_name(args.username)
|
|
if user is None:
|
|
print("Fehler: Benutzer nicht gefunden", file=sys.stderr)
|
|
return 1
|
|
try:
|
|
users.set_password(user.id, _ask_password(), actor=CLI_ACTOR)
|
|
except UserError as exc:
|
|
print(f"Fehler: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(f"Passwort für '{args.username}' gesetzt")
|
|
return 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="checkbot")
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
commands.add_parser("gen-keys")
|
|
create = commands.add_parser("create-admin")
|
|
create.add_argument("username", nargs="?")
|
|
reset = commands.add_parser("set-password")
|
|
reset.add_argument("username")
|
|
messages = commands.add_parser("messages")
|
|
messages.add_argument("--grep", default=None)
|
|
messages.add_argument("--backend", default=None)
|
|
messages.add_argument("--errors", action="store_true")
|
|
messages.add_argument("--since", default=None)
|
|
messages.add_argument("--limit", type=int, default=100)
|
|
args = parser.parse_args(argv)
|
|
handlers = {
|
|
"gen-keys": cmd_gen_keys,
|
|
"create-admin": cmd_create_admin,
|
|
"set-password": cmd_set_password,
|
|
"messages": cmd_messages,
|
|
}
|
|
return handlers[args.command](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|