Blueforce-Linux/scripts/identity/bf-reidentity

#!/usr/bin/env python3
"""Atomically migrate a Blueforce device identity to a new dealer number."""
from __future__ import annotations

import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

IDENTITY_FILE = Path(os.environ.get("BF_IDENTITY_FILE", "/etc/blueforce/dealer-id"))
AUDIT_FILE = Path(os.environ.get("BF_IDENTITY_AUDIT", "/var/log/blueforce-identity.log"))


def run(*args: str) -> None:
    subprocess.run(args, check=True, text=True, capture_output=True)


def valid(value: str) -> bool:
    return len(value) == 8 and value.isdigit()


def audit(message: str) -> None:
    AUDIT_FILE.parent.mkdir(parents=True, exist_ok=True)
    with AUDIT_FILE.open("a", encoding="utf-8") as handle:
        handle.write(message + "\n")
    AUDIT_FILE.chmod(0o600)


def main() -> int:
    if os.geteuid() != 0:
        print("HATA: bayi kimliği geçişi için sudo bf gerekir.", file=sys.stderr)
        return 3
    if len(sys.argv) != 2 or not valid(sys.argv[1]):
        print("Kullanım: sudo bf-reidentity YYYYYYYY", file=sys.stderr)
        return 2
    new_id = sys.argv[1]
    try:
        old_id = IDENTITY_FILE.read_text(encoding="utf-8").strip()
    except OSError as exc:
        print(f"HATA: mevcut bayi kimliği okunamadı: {exc}", file=sys.stderr)
        return 1
    if not valid(old_id) or old_id == new_id:
        print("HATA: yeni kimlik mevcut kimlikten farklı ve 8 haneli olmalıdır.", file=sys.stderr)
        return 2

    old_user = f"bf-{old_id}"
    new_user = f"bf-{new_id}"
    if subprocess.run(("id", "-u", new_user), capture_output=True).returncode == 0:
        print(f"HATA: hedef kullanıcı zaten var: {new_user}", file=sys.stderr)
        return 1
    if subprocess.run(("id", "-u", old_user), capture_output=True).returncode != 0:
        print(f"HATA: mevcut kullanıcı bulunamadı: {old_user}", file=sys.stderr)
        return 1

    old_group = old_user
    old_home = Path(f"/home/{old_user}")
    new_home = Path(f"/home/{new_user}")
    old_hostname = subprocess.run(("hostname",), check=True, capture_output=True, text=True).stdout.strip()
    renamed = False
    group_renamed = False
    moved_sudo = moved_ssh = False
    new_sudo = Path(f"/etc/sudoers.d/{new_user}")
    old_sudo = Path(f"/etc/sudoers.d/{old_user}")
    new_ssh = Path(f"/etc/ssh/sshd_config.d/99-{new_user}.conf")
    old_ssh = Path(f"/etc/ssh/sshd_config.d/99-{old_user}.conf")
    try:
        audit(f"identity-change requested old={old_id} new={new_id}")
        run("usermod", "-l", new_user, old_user)
        renamed = True
        if subprocess.run(("getent", "group", old_group), capture_output=True).returncode == 0:
            run("groupmod", "-n", new_user, old_group)
            group_renamed = True
        run("usermod", "-d", f"/home/{new_user}", "-m", new_user)

        if old_sudo.exists():
            shutil.move(old_sudo, new_sudo)
            moved_sudo = True
            # Scope the rename to full-token occurrences of the old login name (a
            # blind str.replace would also rewrite e.g. bf-11111111-backup).
            sudo_token = re.compile(rf"(?<![\w-]){re.escape(old_user)}(?![\w-])")
            sudo_lines = new_sudo.read_text(encoding="utf-8").splitlines(keepends=True)
            new_sudo.write_text("".join(sudo_token.sub(new_user, line) for line in sudo_lines), encoding="utf-8")
            run("visudo", "-cf", str(new_sudo))

        if old_ssh.exists():
            shutil.move(old_ssh, new_ssh)
            moved_ssh = True
            # Replace only the full AllowUsers token, not substrings of other names.
            ssh_token = re.compile(rf"(?<![\w-]){re.escape(old_user)}(?![\w-])")
            out_lines = []
            for line in new_ssh.read_text(encoding="utf-8").splitlines(keepends=True):
                if line.lstrip().startswith("AllowUsers"):
                    line = ssh_token.sub(new_user, line)
                out_lines.append(line)
            new_ssh.write_text("".join(out_lines), encoding="utf-8")
            run("sshd", "-t")

        tmp = IDENTITY_FILE.with_suffix(".tmp")
        tmp.write_text(new_id + "\n", encoding="utf-8")
        tmp.chmod(0o600)
        os.replace(tmp, IDENTITY_FILE)
        run("hostnamectl", "set-hostname", new_user)
        audit(f"identity-change completed old={old_id} new={new_id} user={new_user}")
        print(f"OK: {old_id} -> {new_id} ({new_user})")
        return 0
    except Exception as exc:
        rollback_errors = []
        try:
            if IDENTITY_FILE.read_text(encoding="utf-8").strip() == new_id:
                IDENTITY_FILE.write_text(old_id + "\n", encoding="utf-8")
                IDENTITY_FILE.chmod(0o600)
            if moved_ssh and new_ssh.exists():
                shutil.move(new_ssh, old_ssh)
            if moved_sudo and new_sudo.exists():
                shutil.move(new_sudo, old_sudo)
            if renamed:
                run("usermod", "-d", str(old_home), "-m", new_user)
            if group_renamed:
                run("groupmod", "-n", old_group, new_user)
            if renamed:
                run("usermod", "-l", old_user, new_user)
            if old_hostname:
                run("hostnamectl", "set-hostname", old_hostname)
        except Exception as rollback_exc:
            rollback_errors.append(str(rollback_exc))
        # A restored config is only trustworthy after it revalidates; and a
        # failed rollback must never be silent.
        try:
            if moved_sudo and old_sudo.exists():
                run("visudo", "-cf", str(old_sudo))
        except Exception as rollback_exc:
            rollback_errors.append(f"sudoers revalidation failed: {rollback_exc}")
        try:
            if moved_ssh and old_ssh.exists():
                run("sshd", "-t")
        except Exception as rollback_exc:
            rollback_errors.append(f"sshd revalidation failed: {rollback_exc}")
        audit(f"identity-change failed old={old_id} new={new_id}: {exc}")
        if rollback_errors:
            print(
                f"HATA: kimlik geçişi geri alınmaya çalışıldı: {exc}; "
                f"geri alma HATALARI: {'; '.join(rollback_errors)}",
                file=sys.stderr,
            )
            return 1
        print(f"HATA: kimlik geçişi geri alınmaya çalışıldı: {exc}", file=sys.stderr)
        return 1


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