#!/usr/bin/python3
"""Read-only, local Linux login/boot assessment. No passwords, hashes or network."""
import argparse
import datetime as dt
import json
import os
from pathlib import Path
import re
import stat
import subprocess
import sys
import tempfile
import uuid

MAX_BYTES = 256 * 1024
COMMAND_TIMEOUT = 4
STATUS = {"observed", "review", "unavailable", "info"}
BIN_DIRS = ("/usr/sbin", "/usr/bin", "/sbin", "/bin")


def clean(value, limit=220):
    value = re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", str(value))
    return "".join(c if c.isprintable() else " " for c in value)[:limit]


def finding(key, title, state, detail, action="", **evidence):
    assert state in STATUS
    return dict(id=key, title=title, status=state, detail=detail,
                action=action, evidence=evidence)


def read_text(path, maximum=MAX_BYTES):
    try:
        with open(path, encoding="utf-8", errors="replace") as handle:
            text = handle.read(maximum + 1)
        return text if len(text) <= maximum else None
    except OSError:
        return None


def executable(name):
    # Never search a writable working directory or the caller's PATH.
    for directory in BIN_DIRS:
        path = Path(directory) / name
        try:
            meta = path.stat()
            if stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and not (meta.st_mode & 0o022) and os.access(path, os.X_OK):
                return str(path)
        except OSError:
            pass
    return None


def command(name, args=(), include_stderr=False):
    binary = executable(name)
    if not binary:
        return None, "", "not available"
    # Commands are fixed read-only queries. DEVNULL prevents prompts. Private
    # temporary files bound returned output without storing it in the report.
    env = {"PATH": ":".join(BIN_DIRS), "LC_ALL": "C", "LANG": "C"}
    try:
        with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err:
            result = subprocess.run([binary, *args], stdin=subprocess.DEVNULL,
                                    stdout=out, stderr=err, timeout=COMMAND_TIMEOUT,
                                    env=env, check=False, close_fds=True)
            if out.tell() > MAX_BYTES or err.tell() > MAX_BYTES:
                return None, "", "output too large"
            out.seek(0)
            text = out.read(MAX_BYTES).decode("utf-8", "replace")
            if include_stderr and result.returncode == 0:
                # LightDM versions may write --show-config to stderr. Only
                # this fixed configuration query opts in; raw text is never
                # included in a report, and both streams retain size limits.
                err.seek(0)
                text += "\n" + err.read(MAX_BYTES).decode("utf-8", "replace")
            return result.returncode, text, "" if result.returncode == 0 else "query not permitted or unsupported"
    except subprocess.TimeoutExpired:
        return None, "", "query timed out"
    except OSError:
        return None, "", "query unavailable"


def account_check(uid, groups):
    admin = bool(set(groups) & {"sudo", "wheel", "admin"})
    if uid == 0:
        return finding("privilege", "Current Linux privileges", "review",
                       "This audit is running as root; applications started as root have system-wide authority.",
                       "Use the normal desktop account for routine work.", uid=uid)
    return finding("privilege", "Current Linux privileges", "review" if admin else "observed",
                   "The current account belongs to an administrative group. Effective sudo permissions were not tested."
                   if admin else "The current account is not root; no common administrative group was observed. Other privileges were not evaluated.",
                   "Use administrator access only for specific maintenance tasks." if admin else "",
                   uid=uid, common_admin_group=admin)


def password_check(output, account, available=True):
    fields = output.strip().split()
    if not available or len(fields) != 7 or fields[0] != account or fields[1] not in {"P", "L", "NP"}:
        return finding("password", "Current account password metadata", "unavailable",
                       "The fixed passwd --status query did not return usable metadata. No password or hash was read by this audit.")
    state = fields[1]
    messages = {
        "P": "A usable password is configured. Its length, uniqueness and strength were not tested.",
        "L": "Password authentication for this account is locked. Other methods such as SSH keys or desktop autologin may still permit access.",
        "NP": "This account has no password configured. Actual access also depends on PAM and display-manager settings.",
    }
    return finding("password", "Current account password metadata", "review" if state == "NP" else "observed",
                   messages[state], "Set a unique account password using the desktop's normal account settings." if state == "NP" else
                   "Never send a password to Garry. This check does not attempt a login or guess passwords.", password_state=state)


def lightdm_autologin(text, available=True):
    if not available or not text:
        return finding("autologin", "Desktop automatic login", "unavailable",
                       "LightDM's effective configuration could not be read. Other display managers are not covered.")
    section = ""
    seats = {}
    for raw in text.splitlines():
        line = raw.strip()
        # --show-config prefixes some effective values with a source letter.
        line = re.sub(r"^[A-Z]\s+", "", line)
        if line.startswith("[") and line.endswith("]"):
            section = line[1:-1]
        elif section.startswith("Seat:") and "=" in line:
            key, value = line.split("=", 1)
            if key.strip() in {"autologin-user", "autologin-guest"}:
                seats.setdefault(section, {})[key.strip()] = value.strip()
    if not seats:
        return finding("autologin", "Desktop automatic login", "unavailable",
                       "No explicit autologin settings were found in LightDM's effective output; no disabled status is inferred.")
    enabled = any(bool(v.get("autologin-user")) or v.get("autologin-guest", "").lower() == "true" for v in seats.values())
    return finding("autologin", "Desktop automatic login", "review" if enabled else "observed",
                   "Automatic desktop login is configured for at least one seat. Someone booting this USB may reach that desktop without an account password."
                   if enabled else "The explicit autologin settings examined were empty or disabled. Display-manager runtime and all seat matching rules were not tested.",
                   "For a private portable desktop, review automatic login and use encrypted persistence; changing login alone does not encrypt stored files." if enabled else "",
                   configured=enabled, scope="LightDM effective configuration")


def configured_autologin_fallback(configs):
    """Positive-risk evidence only; file snippets cannot prove effective state."""
    for text in configs:
        section = ""
        for raw in text.splitlines():
            line = raw.strip()
            if line.startswith("[") and line.endswith("]"):
                section = line[1:-1]
            elif (section.startswith("Seat:") or section == "SeatDefaults") and "=" in line:
                key, value = line.split("=", 1)
                if (key.strip() == "autologin-user" and value.strip()) or (
                        key.strip() == "autologin-guest" and value.strip().lower() == "true"):
                    return finding("autologin", "Desktop automatic login", "review",
                                   "A readable LightDM configuration file contains an automatic-login setting. The merged effective configuration was unavailable; another file may override it.",
                                   "Treat unattended desktop access as a risk to review. A login password alone does not encrypt portable files.",
                                   configured_entry_observed=True, effective_state="unavailable",
                                   scope="readable LightDM configuration entries only")
    return finding("autologin", "Desktop automatic login", "unavailable",
                   "LightDM's effective configuration was unavailable, and readable configuration entries did not establish automatic login. This does not prove automatic login is disabled.")


def lightdm_config_texts():
    """Only known LightDM configuration locations; no symlink file reads."""
    paths = [Path("/etc/lightdm/lightdm.conf")]
    for directory in ("/usr/share/lightdm/lightdm.conf.d", "/usr/local/share/lightdm/lightdm.conf.d", "/etc/lightdm/lightdm.conf.d"):
        try:
            paths.extend(sorted(Path(directory).glob("*.conf"))[:40])
        except OSError:
            pass
    texts = []
    for path in paths[:121]:
        try:
            meta = path.lstat()
            if not stat.S_ISREG(meta.st_mode):
                continue
            fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
            with os.fdopen(fd, "r", encoding="utf-8", errors="replace") as handle:
                text = handle.read(65537)
            if len(text) <= 65536:
                texts.append(text)
        except OSError:
            pass
    return texts


def secure_boot_check(efi_present, raw):
    if not efi_present:
        return finding("secure_boot", "UEFI Secure Boot", "unavailable",
                       "UEFI runtime information is absent. This may be a legacy BIOS boot or a restricted environment; firmware settings were not changed.")
    if raw is None or len(raw) != 5 or raw[4] not in (0, 1):
        return finding("secure_boot", "UEFI Secure Boot", "unavailable",
                       "The firmware SecureBoot variable was unavailable or unexpected. No enabled status is inferred.")
    enabled = raw[4] == 1
    return finding("secure_boot", "UEFI Secure Boot", "observed" if enabled else "review",
                   "Firmware reports Secure Boot enabled. This does not establish that the desktop, persistence or passwords are safe."
                   if enabled else "Firmware reports Secure Boot disabled for this boot. USB image compatibility must be verified before changing it.",
                   "Review firmware protections and trusted boot media. This audit does not modify firmware.", enabled=enabled)


def unescape_mount(value):
    return re.sub(r"\\([0-7]{3})", lambda match: chr(int(match.group(1), 8)), value)


def parse_mountinfo(text):
    result = []
    for line in (text or "").splitlines():
        try:
            before, after = line.split(" - ", 1)
            left, right = before.split(), after.split()
            if len(left) < 6 or len(right) < 3 or not re.fullmatch(r"\d+:\d+", left[2]):
                continue
            result.append(dict(device=left[2], target=unescape_mount(left[4]), options=left[5],
                               fstype=right[0], source=unescape_mount(right[1]), super_options=right[2]))
        except (ValueError, IndexError):
            continue
    return result


def mount_at(mounts, path):
    return max((m for m in mounts if path == m["target"] or path.startswith(m["target"].rstrip("/") + "/")),
               key=lambda m: len(m["target"]), default=None)


def storage_check(mounts, cmdline):
    root = next((m for m in mounts if m["target"] == "/"), None)
    live = bool(re.search(r"(?:^|\s)boot=live(?:\s|$)", cmdline or "")) or bool(root and root["fstype"] in {"overlay", "aufs"})
    if not root:
        return finding("storage", "Live system and persistence", "unavailable", "The active root mount could not be identified."), None
    if root["fstype"] == "overlay":
        upper = next((unescape_mount(v[9:]) for v in root["super_options"].split(",") if v.startswith("upperdir=")), None)
        backing = mount_at(mounts, upper) if upper else None
        if not backing or backing is root:
            return finding("storage", "Live system and persistence", "unavailable",
                           "A live/overlay root was observed, but its writable backing could not be identified. Full-system persistence is unverified.", live_or_overlay=True), None
        if backing["fstype"] in {"tmpfs", "ramfs", "overlay", "aufs"}:
            return finding("storage", "Live system and persistence", "review",
                           "The current root overlay is backed by temporary storage. System changes are not shown to survive reboot; a separate home-only persistent mount is outside this check.",
                           "Save important files before shutdown. Full-system persistence must be established before relying on persistent settings.", mode="temporary", live_or_overlay=True), None
        if backing["source"].startswith("/dev/"):
            writable = "rw" in root["options"].split(",") and "rw" in backing["options"].split(",")
            return finding("storage", "Live system and persistence", "observed" if writable else "review",
                           "The root overlay has disk-backed storage. It is writable, but reboot survival was not tested." if writable else
                           "The root overlay has disk-backed storage that is not shown writable. New settings may not survive.",
                           mode="full-system-backing", live_or_overlay=True, writable=writable, filesystem=backing["fstype"]), backing
    if not live and root["source"].startswith("/dev/"):
        return finding("storage", "Live system and persistence", "observed",
                       "The current Linux root is mounted from a block device and does not match the expected live overlay layout.",
                       mode="direct-root", live_or_overlay=False, filesystem=root["fstype"]), root
    return finding("storage", "Live system and persistence", "unavailable",
                   "This mount layout is not supported for proving full-system persistence.", live_or_overlay=live), None


def encryption_check(backing, sysroot=Path("/sys")):
    if not backing:
        return finding("encryption", "Current Linux storage encryption", "unavailable", "No mounted persistent backing was identified for an encryption check.")
    if backing["fstype"] not in {"ext2", "ext3", "ext4", "xfs", "btrfs", "f2fs"}:
        return finding("encryption", "Current Linux storage encryption", "unavailable", "The backing filesystem is outside this check's supported scope.")
    start = sysroot / "dev/block" / backing["device"]
    pending, seen = [start], set()
    luks, unknown = False, False
    try:
        while pending and len(seen) < 16:
            device = pending.pop().resolve(strict=True)
            if device in seen:
                continue
            seen.add(device)
            uuid_path = device / "dm/uuid"
            if uuid_path.exists():
                value = read_text(uuid_path, 4096)
                if value is None:
                    unknown = True
                elif value.startswith("CRYPT-LUKS"):
                    luks = True
                elif value.startswith("CRYPT-"):
                    unknown = True
            slaves = device / "slaves"
            if slaves.exists():
                pending.extend(list(slaves.iterdir())[:16])
        unknown = unknown or bool(pending)
    except (OSError, RuntimeError):
        unknown = True
    if luks:
        return finding("encryption", "Current Linux storage encryption", "observed",
                       "A LUKS device-mapper layer was observed beneath the current mounted Linux backing. This does not verify passphrase strength or cover USB boot files, other disks or Windows.", luks_layer_observed=True)
    if unknown or not seen:
        return finding("encryption", "Current Linux storage encryption", "unavailable",
                       "The mounted storage's device chain could not be fully examined. No encryption verdict is available.")
    return finding("encryption", "Current Linux storage encryption", "review",
                   "No LUKS layer was observed beneath the mounted Linux backing. Hardware encryption and other protection mechanisms were not assessed.",
                   "Use a separately verified encrypted-persistence setup for portable private data. A login password alone does not establish disk encryption.", luks_layer_observed=False)


def ssh_check(text, available=True, service_state="unknown"):
    allowed = {"passwordauthentication", "kbdinteractiveauthentication", "permitemptypasswords", "permitrootlogin", "authenticationmethods"}
    values = {}
    for line in text.splitlines():
        fields = line.split(None, 1)
        if len(fields) == 2 and fields[0] in allowed:
            values[fields[0]] = clean(fields[1], 120)
    required = {"passwordauthentication", "kbdinteractiveauthentication", "permitemptypasswords", "permitrootlogin"}
    binary_keys = required - {"permitrootlogin"}
    valid_values = all(values.get(key) in {"yes", "no"} for key in binary_keys) and values.get("permitrootlogin") in {
        "yes", "no", "prohibit-password", "without-password", "forced-commands-only"}
    if not available or not required <= values.keys() or not valid_values:
        return finding("ssh", "Local SSH authentication configuration", "unavailable",
                       "OpenSSH's effective base configuration was unavailable. No server was started and no login was attempted.", service_state=service_state)
    passwords = values["passwordauthentication"] == "yes" or values.get("kbdinteractiveauthentication") == "yes"
    risky = passwords or values["permitemptypasswords"] == "yes" or values["permitrootlogin"] == "yes"
    return finding("ssh", "Local SSH authentication configuration", "review" if risky else "observed",
                   "The effective base configuration permits a password-capable authentication method or a setting requiring review."
                   if risky else "No password-capable method or unrestricted root login was observed in the base configuration.",
                   "Match rules and daemon command-line overrides may differ. Network reachability, actual login policy and keys were not tested; this is not an exposure verdict.",
                   service_state=service_state, scope="sshd -G base configuration", **values)


def collect():
    import grp
    import pwd
    account = pwd.getpwuid(os.getuid()).pw_name
    groups = []
    for gid in set(os.getgroups()) | {os.getgid()}:
        try:
            groups.append(grp.getgrgid(gid).gr_name)
        except KeyError:
            pass
    findings = [account_check(os.geteuid(), groups)]
    rc, text, _ = command("passwd", ("--status", "--", account))
    findings.append(password_check(text, account, rc == 0))
    rc, text, _ = command("lightdm", ("--show-config",), include_stderr=True)
    autologin = lightdm_autologin(text, rc == 0)
    if autologin["status"] == "unavailable":
        autologin = configured_autologin_fallback(lightdm_config_texts())
    findings.append(autologin)
    efi = Path("/sys/firmware/efi")
    raw = None
    try:
        variable = efi / "efivars/SecureBoot-8be4df61-93ca-11d2-aa0d-00e098032b8c"
        with variable.open("rb") as handle:
            raw = handle.read(6)
    except OSError:
        pass
    findings.append(secure_boot_check(efi.exists(), raw))
    mounts = parse_mountinfo(read_text("/proc/self/mountinfo"))
    storage, backing = storage_check(mounts, read_text("/proc/cmdline", 16384))
    findings.extend((storage, encryption_check(backing)))
    rc, service, _ = command("systemctl", ("show", "ssh.service", "--property=ActiveState", "--value", "--no-pager"))
    service = service.strip() if rc == 0 and service.strip() in {"active", "inactive", "activating", "deactivating", "failed"} else "unknown"
    rc, text, _ = command("sshd", ("-G",))
    findings.append(ssh_check(text, rc == 0, service))
    findings.append(finding("windows", "Windows account security", "unavailable",
                            "Booting this Linux USB cannot establish Windows password, Windows Hello, account-lockout or BitLocker security. No Windows volume was opened.",
                            "Run the separate Windows security report while signed into the Windows installation you own."))
    return {"schema_version": 1, "tool": "Garry local Linux login security", "version": "20260923.2",
            "observed_at": dt.datetime.now(dt.timezone.utc).isoformat(), "scope": "current booted Linux system only",
            "overall": "assessment only; no overall pass or guarantee", "findings": findings,
            "limits": ["No password strength test, password guessing, credential extraction, hashes, network scans or changes.",
                       "No other operating system or user's account was assessed. Unavailable does not mean safe.",
                       "Firmware, mount and configuration observations do not prove successful login protections or reboot persistence."]}


def render(report):
    lines = ["GARRY | LOCAL LOGIN SECURITY", report["observed_at"], "Scope: " + report["scope"],
             "Read-only assessment. There is no overall 'secure' score.", ""]
    for item in report["findings"]:
        lines.extend([f'[{item["status"].upper()}] {item["title"]}', item["detail"]])
        if item["action"]:
            lines.append("Next: " + item["action"])
        if item["evidence"]:
            lines.append("Evidence: " + json.dumps(item["evidence"], ensure_ascii=True, sort_keys=True))
        lines.append("")
    lines.extend(["Limits:", *["- " + value for value in report["limits"]]])
    return "\n".join(lines) + "\n"


def save_private(report, directory):
    directory = Path(directory).expanduser()
    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
    metadata = directory.lstat()
    if not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.getuid() or metadata.st_mode & 0o077:
        raise ValueError("Report directory must be a real private directory owned by the current user (mode 0700).")
    basename = "login-security-" + dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ-") + uuid.uuid4().hex[:8]
    paths = []
    for extension, content in (("txt", render(report)), ("json", json.dumps(report, indent=2) + "\n")):
        path = directory / (basename + "." + extension)
        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(content)
        paths.append(str(path))
    return paths


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--json", action="store_true", help="Print structured report rather than human-readable text")
    parser.add_argument("--save", nargs="?", const=str(Path.home() / ".local/state/garry/login-security"),
                        metavar="PRIVATE_DIRECTORY", help="Also save private text/JSON reports (default: current user's local state directory)")
    args = parser.parse_args()
    if sys.platform != "linux":
        parser.exit(2, "This tool assesses only the running Linux system. Use the Windows kit inside Windows.\n")
    report = collect()
    if args.save:
        try:
            report["saved_reports"] = save_private(report, args.save)
        except (OSError, ValueError):
            parser.exit(1, "Could not save reports in a private directory. No system settings were changed.\n")
    print(json.dumps(report, indent=2) if args.json else render(report), end="\n" if args.json else "")
    if args.save and not args.json:
        print("Private reports:\n" + "\n".join(report["saved_reports"]))
    return 0


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