ops: add auth activity and session views

deploy/systemd/bin/metinctl.in

@@ -51,6 +51,20 @@ def parse_args() -> argparse.Namespace:
     summary_parser.add_argument("--include-smoke", action="store_true", help="Include smoke-test logins in auth summary")
     summary_parser.add_argument("--json", action="store_true", help="Print raw JSON")
 
+    auth_activity = subparsers.add_parser("auth-activity", help="Show recent auth success/failure activity")
+    auth_activity.add_argument("--hours", type=int, default=24, help="How many hours back to inspect")
+    auth_activity.add_argument("--limit", type=int, default=30, help="Maximum events to show")
+    auth_activity.add_argument("--status", choices=("all", "success", "failure"), default="all", help="Filter by auth result")
+    auth_activity.add_argument("--include-smoke", action="store_true", help="Include smoke-test logins")
+    auth_activity.add_argument("--json", action="store_true", help="Print raw JSON")
+
+    sessions = subparsers.add_parser("sessions", help="Show recent login sessions from loginlog2")
+    sessions.add_argument("--hours", type=int, default=24, help="How many hours back to inspect")
+    sessions.add_argument("--limit", type=int, default=20, help="Maximum sessions to show")
+    sessions.add_argument("--active-only", action="store_true", help="Show only sessions without logout_time")
+    sessions.add_argument("--include-orphans", action="store_true", help="Include rows whose account login no longer exists")
+    sessions.add_argument("--json", action="store_true", help="Print raw JSON")
+
     status_parser = subparsers.add_parser("status", help="Show current unit state")
     status_parser.add_argument("target", nargs="?", default="all", help="stack, db, auth, game, channel:<id>, instance:<name>")
 
@@ -325,6 +339,69 @@ def summarize_auth_activity(hours: int, include_smoke: bool) -> dict[str, object
     }
 
 
+def filter_auth_events(hours: int, include_smoke: bool, status: str) -> list[dict[str, object]]:
+    events = load_auth_activity(hours)
+    filtered = [event for event in events if include_smoke or not event["smoke"]]
+    if status != "all":
+        filtered = [event for event in filtered if event["status"] == status]
+    return filtered
+
+
+def run_mariadb_query(query: str) -> list[list[str]]:
+    completed = run(["mariadb", "-N", "-B", "-e", query], require_root=True, capture_output=True)
+    rows: list[list[str]] = []
+    for line in completed.stdout.splitlines():
+        if not line.strip():
+            continue
+        rows.append(line.split("\t"))
+    return rows
+
+
+def fetch_recent_sessions(hours: int, limit: int, active_only: bool, include_orphans: bool) -> list[dict[str, str]]:
+    where_clauses = [f"l.login_time >= NOW() - INTERVAL {int(hours)} HOUR"]
+    if active_only:
+        where_clauses.append("l.logout_time IS NULL")
+    if not include_orphans:
+        where_clauses.append("a.login IS NOT NULL")
+
+    query = f"""
+SELECT
+  DATE_FORMAT(l.login_time, '%Y-%m-%d %H:%i:%s'),
+  COALESCE(DATE_FORMAT(l.logout_time, '%Y-%m-%d %H:%i:%s'), ''),
+  l.type,
+  COALESCE(a.login, ''),
+  l.account_id,
+  l.pid,
+  COALESCE(INET_NTOA(l.ip), ''),
+  COALESCE(l.client_version, '')
+FROM log.loginlog2 l
+LEFT JOIN account.account a ON a.id = l.account_id
+WHERE {' AND '.join(where_clauses)}
+ORDER BY l.id DESC
+LIMIT {int(limit)}
+""".strip()
+
+    entries: list[dict[str, str]] = []
+    for row in run_mariadb_query(query):
+        while len(row) < 8:
+            row.append("")
+        login_time, logout_time, raw_type, login, account_id, pid, ip, client_version = row[:8]
+        entries.append(
+            {
+                "login_time": login_time,
+                "logout_time": logout_time,
+                "raw_type": raw_type,
+                "session_state": "open" if not logout_time else "closed",
+                "login": login or f"<missing:{account_id}>",
+                "account_id": account_id,
+                "pid": pid,
+                "ip": ip or "-",
+                "client_version": client_version or "-",
+            }
+        )
+    return entries
+
+
 def live_ports() -> set[int]:
     if shutil.which("ss") is None:
         return set()
@@ -483,6 +560,49 @@ def print_summary(hours: int, include_smoke: bool, as_json: bool) -> int:
     return 0
 
 
+def print_auth_activity(hours: int, limit: int, status: str, include_smoke: bool, as_json: bool) -> int:
+    events = filter_auth_events(hours, include_smoke, status)
+    events = events[-limit:]
+    payload = {
+        "window_hours": hours,
+        "limit": limit,
+        "status": status,
+        "include_smoke": include_smoke,
+        "count": len(events),
+        "entries": [
+            {
+                "time": event["time"].strftime("%Y-%m-%d %H:%M:%S"),
+                "status": event["status"],
+                "login": event["login"],
+                "ip": event["ip"],
+                "reason": event["reason"],
+            }
+            for event in events
+        ],
+    }
+
+    if as_json:
+        print(json.dumps(payload, indent=2))
+        return 0
+
+    if not events:
+        print(f"No auth activity in the last {hours}h for status={status}.")
+        return 0
+
+    rows = [
+        [
+            event["time"].strftime("%Y-%m-%d %H:%M:%S"),
+            str(event["status"]),
+            str(event["login"]),
+            str(event["ip"]),
+            str(event["reason"]),
+        ]
+        for event in events
+    ]
+    print_table(["time", "status", "login", "ip", "reason"], rows)
+    return 0
+
+
 def resolve_target_units(target: str) -> list[str]:
     normalized = target.strip().lower()
 
@@ -642,6 +762,41 @@ def print_auth_failures(hours: int, limit: int, include_smoke: bool, as_json: bo
     return 0
 
 
+def print_sessions(hours: int, limit: int, active_only: bool, include_orphans: bool, as_json: bool) -> int:
+    entries = fetch_recent_sessions(hours, limit, active_only, include_orphans)
+    payload = {
+        "window_hours": hours,
+        "limit": limit,
+        "active_only": active_only,
+        "include_orphans": include_orphans,
+        "count": len(entries),
+        "entries": entries,
+    }
+
+    if as_json:
+        print(json.dumps(payload, indent=2))
+        return 0
+
+    if not entries:
+        print(f"No sessions in the last {hours}h.")
+        return 0
+
+    rows = [
+        [
+            entry["login_time"],
+            entry["logout_time"] or "-",
+            entry["session_state"],
+            entry["login"],
+            entry["account_id"],
+            entry["pid"],
+            entry["ip"],
+        ]
+        for entry in entries
+    ]
+    print_table(["login_time", "logout_time", "state", "login", "account", "pid", "ip"], rows)
+    return 0
+
+
 def run_healthcheck(mode: str) -> int:
     if not HEALTHCHECK_PATH.exists():
         raise SystemExit(f"Missing healthcheck wrapper: {HEALTHCHECK_PATH}")
@@ -696,6 +851,8 @@ def main() -> int:
         return print_units()
     if args.command == "summary":
         return print_summary(args.hours, args.include_smoke, args.json)
+    if args.command == "auth-activity":
+        return print_auth_activity(args.hours, args.limit, args.status, args.include_smoke, args.json)
     if args.command == "status":
         return print_status(args.target)
     if args.command == "ports":
@@ -706,6 +863,8 @@ def main() -> int:
         return print_incidents(args.limit)
     if args.command == "auth-failures":
         return print_auth_failures(args.hours, args.limit, args.include_smoke, args.json)
+    if args.command == "sessions":
+        return print_sessions(args.hours, args.limit, args.active_only, args.include_orphans, args.json)
     if args.command in {"start", "stop", "restart"}:
         return run_unit_action(args.command, args.target)
     if args.command == "logs":