List detached sessions of mosh

list-mosh-detached.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
List detached mosh PIDs by scanning utmp, emulating mosh-server's warn_unattached() logic.
- Linux + glibc 前提（libc.so.6 の utmpx 構造体）
- "mosh [PID]" 形式の ut_host を「detached」とみなして抽出
"""

import os
import re
import pwd
import argparse
from ctypes import CDLL, Structure, POINTER, c_short, c_int, c_long, c_char, c_char_p

# --- glibc の utmpx 構造体（/usr/include/bits/utmp.h 相当） ---
class ExitStatus(Structure):
    _fields_ = [
        ("e_termination", c_short),
        ("e_exit", c_short),
    ]

class TimeVal(Structure):
    _fields_ = [
        ("tv_sec", c_long),
        ("tv_usec", c_long),
    ]

class Utmpx(Structure):
    _fields_ = [
        ("ut_type", c_short),
        ("ut_pid", c_int),
        ("ut_line", c_char * 32),
        ("ut_id", c_char * 4),
        ("ut_user", c_char * 32),
        ("ut_host", c_char * 256),
        ("ut_exit", ExitStatus),
        ("ut_session", c_long),
        ("ut_tv", TimeVal),
        ("ut_addr_v6", c_int * 4),
        ("__unused", c_char * 20),
    ]

# ut_type 定数（man 5 utmp）
EMPTY, RUN_LVL, BOOT_TIME, NEW_TIME, OLD_TIME, INIT_PROCESS, LOGIN_PROCESS, USER_PROCESS, DEAD_PROCESS, ACCOUNTING = range(0, 10)

MOSH_DETACHED_RE = re.compile(r"^mosh \[(\d+)\]$")  # detached
MOSH_ATTACHED_RE = re.compile(r".+\s+via mosh \[(\d+)\]$")  # attached（参考：除外用）

def current_username() -> str:
    try:
        return pwd.getpwuid(os.getuid()).pw_name
    except Exception:
        # まれに getlogin() のほうが取れる環境もある
        return os.environ.get("LOGNAME") or os.environ.get("USER") or ""

def list_detached_pids(user: str | None = None) -> list[int]:
    """
    utmpx を走査して、指定ユーザの「detached mosh」PID を列挙。
    """
    target_user = user or current_username()
    libc = CDLL("libc.so.6", use_errno=True)

    # 関数プロトタイプ
    libc.setutxent.argtypes = []
    libc.setutxent.restype = None

    libc.getutxent.argtypes = []
    libc.getutxent.restype = POINTER(Utmpx)

    libc.endutxent.argtypes = []
    libc.endutxent.restype = None

    detached_pids: list[int] = []
    libc.setutxent()
    try:
        while True:
            entry_p = libc.getutxent()
            if not entry_p:
                break
            u = entry_p.contents
            if u.ut_type != USER_PROCESS:
                continue

            user_s = bytes(u.ut_user).split(b"\x00", 1)[0].decode(errors="ignore")
            host_s = bytes(u.ut_host).split(b"\x00", 1)[0].decode(errors="ignore").strip()

            if target_user and user_s != target_user:
                continue

            # attached は除外（"IP via mosh [PID]"）
            if MOSH_ATTACHED_RE.search(host_s):
                continue

            m = MOSH_DETACHED_RE.match(host_s)
            if m:
                try:
                    pid = int(m.group(1))
                    detached_pids.append(pid)
                except ValueError:
                    pass
    finally:
        libc.endutxent()

    # 重複排除＋整列（見やすさ）
    return sorted(set(detached_pids))

def main():
    ap = argparse.ArgumentParser(description="Enumerate detached mosh PIDs by scanning utmp.")
    ap.add_argument("-u", "--user", help="target username (default: current user)", default=None)
    ap.add_argument("--print-mosh-style", action="store_true",
                    help='Print in the same style as mosh-server ("- mosh [PID]")')
    args = ap.parse_args()

    pids = list_detached_pids(args.user)

    if args.print_mosh_style:
        if not pids:
            print("No detached Mosh sessions found.")
            return
        if len(pids) == 1:
            print(f"Mosh: You have a detached Mosh session on this server (mosh [{pids[0]}]).")
        else:
            print(f"Mosh: You have {len(pids)} detached Mosh sessions on this server, with PIDs:")
            for pid in pids:
                print(f"        - mosh [{pid}]")
    else:
        # プレーンな PID 群（スクリプトから扱いやすい）
        print(" ".join(str(pid) for pid in pids))

if __name__ == "__main__":
    main()