//! Windows user administration (port of `users/windowsUserAdminService.js`).
//!
//! Every operation runs the fixed `remoteUsers.ps1` through the Windows executor. Built-in
//! accounts, domain/cloud identities and the session account are protected. sshd_config
//! changes are staged (`sshd -t`, `sshd -T -f candidate`), committed through a SYSTEM
//! scheduled task that arms the rollback before restarting sshd, and kept only after a NEW
//! SSH login confirmed them.

use super::common::*;
use super::keys;
use super::policy::{self, AccessContext, Effective, Policies, Policy, VerifyOptions};
use super::windows_policy::{self as winpolicy, ManagedPolicies};
use crate::core::error::CoreResult;
use crate::core::logging::Logger;
use crate::core::ssh::session::SessionManager;
use crate::core::ssh::windows_exec::{bootstrap_input, encoded_command, parse_windows_output, WindowsExecutor, BOOTSTRAP};
use base64::engine::general_purpose::STANDARD as B64;
use base64::Engine;
use once_cell::sync::Lazy;
use rand::RngCore;
use regex::Regex;
use serde_json::{json, Map, Value};
use std::sync::Arc;
use std::time::Duration;

pub const REMOTE_SCRIPT: &str = include_str!("../../../resources/remote/windows/remoteUsers.ps1");
pub const ADMINS_SID: &str = "S-1-5-32-544";
pub const APPLY_TIMEOUT_SECONDS: u64 = 120;
static LOCAL_SID: Lazy<Regex> = Lazy::new(|| Regex::new(r"^S-1-5-21-\d+-\d+-\d+-\d+$").unwrap());
static GROUP_SID: Lazy<Regex> = Lazy::new(|| Regex::new(r"^S-1-5-(32-\d+|21-\d+-\d+-\d+-\d+)$").unwrap());
static NAME: Lazy<Regex> = Lazy::new(|| Regex::new(r"^[A-Za-z0-9_][A-Za-z0-9._ -]{0,19}$").unwrap());

fn action_name(action: &str) -> Option<&'static str> {
    Some(match action {
        "enable" | "unexpire" => "enable",
        "disable" | "expire" => "disable",
        "setPassword" => "password",
        "setFullName" | "setComment" => "fullName",
        "setDescription" => "description",
        "addGroup" => "group-add",
        "removeGroup" => "group-remove",
        "grantAdmin" => "admin-grant",
        "revokeAdmin" => "admin-revoke",
        _ => return None,
    })
}

pub fn validate_windows_name(value: &Value) -> CoreResult<String> {
    let name = js_string(value).trim().to_string();
    if !NAME.is_match(&name) || name.ends_with('.') || name.ends_with(' ') {
        return Err(fail("USERS_NAME_INVALID", "Windows account names may contain letters, digits, space, '.', '_' or '-' (max. 20) and must not end with a dot or space."));
    }
    Ok(name)
}

pub fn validate_sid(value: &Value, pattern: &Regex, code: &str) -> CoreResult<String> {
    let sid = js_string(value).trim().to_uppercase();
    if !pattern.is_match(&sid) {
        return Err(fail(code, "Invalid security identifier (SID)."));
    }
    Ok(sid)
}

fn validate_password(value: &Value) -> CoreResult<String> {
    let password = js_string(value);
    if password.is_empty() || password.encode_utf16().count() > 256 || password.contains(['\0', '\r', '\n']) {
        return Err(fail("USERS_PASSWORD_INVALID", "The password must contain 1-256 characters and no line breaks."));
    }
    Ok(password)
}

fn validate_text(value: &Value, limit: usize, code: &str) -> CoreResult<String> {
    let text = js_string(value).trim().to_string();
    if text.encode_utf16().count() > limit || text.chars().any(|c| (c as u32) < 0x20 || c as u32 == 0x7f) {
        return Err(fail(code, &format!("The text may contain up to {limit} characters without control characters.")));
    }
    Ok(text)
}

pub fn privilege_from_token(token: &Value, system: &Value) -> Value {
    let elevated = token["elevated"] == true;
    let admin_member = token["adminMember"] == true;
    let mode = if elevated { "windows-elevated" } else if admin_member { "windows-admin-unelevated" } else { "windows-standard" };
    let groups: Vec<String> = token["groupNames"].as_array().map(|g| g.iter().take(256).map(|n| js_string(n).to_lowercase()).collect()).unwrap_or_default();
    json!({
        "platform": "windows",
        "mode": mode,
        "verified": true,
        "username": js_string(&token["name"]),
        "sid": js_string(&token["sid"]),
        "integrity": token["integrity"].as_str().map(str::to_string).unwrap_or_else(|| if token["integrity"].is_null() { "unknown".into() } else { js_string(&token["integrity"]) }),
        "adminMember": admin_member,
        "elevated": elevated,
        "sudoInstalled": false,
        "groups": groups,
        "clientAddress": js_string(&system["sshClient"]).trim().split_whitespace().next().unwrap_or(""),
        "message": "",
        "checkedAt": now_iso()
    })
}

/// The account name used in sshd_config and in sshd -T for a local user ("codex", not "HOST\\codex").
fn account_name(token_name: &str) -> String {
    token_name.rsplit('\\').next().unwrap_or("").to_string()
}

fn decode_config(ssh: &Value) -> String {
    ssh["configData"].as_str().filter(|s| !s.is_empty()).map(|d| String::from_utf8_lossy(&B64.decode(d).unwrap_or_default()).to_string()).unwrap_or_default()
}

/// Windows key file with internal raw bytes.
pub struct WinKeyFile {
    pub path: String,
    pub kind: String,
    pub shared: bool,
    pub manageable: bool,
    pub entries: Vec<keys::KeyEntry>,
    pub raw: Vec<u8>,
    pub public: Value,
    pub acl_warnings: Vec<String>,
}

pub struct WinDetail {
    pub public: Value,
    pub key_files: Vec<WinKeyFile>,
    pub config_text: String,
    pub config_sha256: Value,
    pub managed: Option<ManagedPolicies>,
    pub addresses: Addresses,
    pub groups: Vec<String>,
    pub privilege: Value,
    pub sid: String,
    pub username: String,
    pub self_account: bool,
}

pub struct WindowsUsers {
    pub sessions: Arc<SessionManager>,
    pub executor: Arc<WindowsExecutor>,
    pub logger: Arc<Logger>,
}

impl WindowsUsers {
    async fn run(&self, session_id: &str, args: Value, purpose: &str, timeout_ms: u64) -> CoreResult<Value> {
        self.executor.run(session_id, REMOTE_SCRIPT, &args, Duration::from_millis(timeout_ms), purpose, "USERS_").await
    }

    pub async fn privileges(&self, session_id: &str) -> CoreResult<Value> {
        let probe = self.run(session_id, json!({ "op": "probe" }), "users.windows.probe", 60_000).await?;
        Ok(privilege_from_token(&probe["token"], &probe["system"]))
    }

    pub async fn inventory(&self, session_id: &str, platform: &Value) -> CoreResult<Value> {
        let raw = self.run(session_id, json!({ "op": "inventory" }), "users.windows.inventory", 120_000).await?;
        Ok(self.map_inventory(session_id, &raw, platform))
    }

    fn map_inventory(&self, session_id: &str, raw: &Value, platform: &Value) -> Value {
        let privilege = privilege_from_token(&raw["token"], &raw["system"]);
        let config = decode_config(&raw["ssh"]);
        let managed = (raw["ssh"]["configReadable"] == true).then(|| winpolicy::parse_managed_policies(&config));
        let raw_users = raw["users"].as_array().cloned().unwrap_or_default();
        let machine_prefix = raw_users.iter().find(|u| LOCAL_SID.is_match(u["sid"].as_str().unwrap_or(""))).and_then(|u| u["sid"].as_str()).map(|sid| sid.rsplit_once('-').map(|(p, _)| p.to_string()).unwrap_or_default()).unwrap_or_default();
        let groups: Vec<Value> = raw["groups"]
            .as_array()
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .map(|g| {
                let members: Vec<Value> = g["members"].as_array().cloned().unwrap_or_default().into_iter().map(|m| json!({ "sid": m["sid"], "name": clean_text(&js_string(&m["name"]), 120), "source": m["source"], "objectClass": m["objectClass"] })).collect();
                json!({ "name": js_string(&g["name"]), "sid": js_string(&g["sid"]), "description": clean_text(&js_string(&g["description"]), 200), "members": members })
            })
            .collect();
        let mut users: Vec<Value> = raw_users
            .iter()
            .map(|user| {
                let sid = js_string(&user["sid"]);
                let member_of: Vec<&Value> = groups.iter().filter(|g| g["members"].as_array().is_some_and(|m| m.iter().any(|m| m["sid"] == sid.as_str()))).collect();
                let rid: i64 = sid.rsplit('-').next().and_then(|r| r.parse().ok()).unwrap_or(0);
                let builtin = rid < 1000;
                let name = winpolicy::ssh_name(&js_string(&user["name"]));
                let ssh_policy = match (&name, &managed) {
                    (Some(n), Some(m)) => m.policies.get(n).map(Policy::to_value).unwrap_or(Value::Null),
                    _ => Value::Null,
                };
                json!({
                    "platform": "windows",
                    "username": user["name"],
                    "sid": sid,
                    "rid": rid,
                    "enabled": user["enabled"] == true,
                    "fullName": clean_text(&js_string(&user["fullName"]), 256),
                    "comment": clean_text(&js_string(&user["fullName"]), 256),
                    "description": clean_text(&js_string(&user["description"]), 256),
                    "source": user["source"],
                    "lastLogon": user["lastLogon"],
                    "passwordLastSet": user["passwordLastSet"],
                    "passwordRequired": user["passwordRequired"] == true,
                    "passwordExpires": user["passwordExpires"],
                    "accountExpires": user["accountExpires"],
                    "admin": member_of.iter().any(|g| g["sid"] == ADMINS_SID),
                    "groups": member_of.iter().map(|g| g["name"].clone()).collect::<Vec<_>>(),
                    "groupSids": member_of.iter().map(|g| g["sid"].clone()).collect::<Vec<_>>(),
                    "system": builtin,
                    "builtin": builtin,
                    "protection": if builtin { json!("builtin") } else if sid == privilege["sid"].as_str().unwrap_or("") { json!("session") } else { Value::Null },
                    "sshName": name,
                    "sshPolicy": ssh_policy
                })
            })
            .collect();
        users.sort_by(|a, b| (a["system"] == true).cmp(&(b["system"] == true)).then_with(|| policy::locale_cmp(a["username"].as_str().unwrap_or(""), b["username"].as_str().unwrap_or(""))));
        // Members that are not local accounts of this machine (domain, Microsoft Entra, other machines) are read-only.
        let mut external: Vec<Value> = Vec::new();
        for group in &groups {
            for member in group["members"].as_array().cloned().unwrap_or_default() {
                let sid = js_string(&member["sid"]);
                if !sid.starts_with("S-1-5-21-") && !sid.starts_with("S-1-12-1-") {
                    continue;
                }
                if !machine_prefix.is_empty() && sid.starts_with(&format!("{machine_prefix}-")) {
                    continue;
                }
                match external.iter_mut().find(|e| e["sid"] == sid.as_str()) {
                    Some(entry) => entry["groups"].as_array_mut().unwrap().push(group["name"].clone()),
                    None => external.push(json!({ "sid": sid, "name": member["name"], "source": member["source"], "groups": [group["name"]] })),
                }
            }
        }
        let system = &raw["system"];
        let display = js_string(&system["displayVersion"]);
        json!({
            "platform": "windows",
            "sessionId": session_id,
            "privileged": privilege["elevated"],
            "privilege": privilege,
            "server": {
                "os": format!("{} ({}{})", js_string(&system["caption"]), js_string(&system["version"]), if display.is_empty() { String::new() } else { format!(", {display}") }),
                "osId": "windows",
                "installationType": system["installationType"],
                "serverCore": js_string(&system["installationType"]).to_lowercase().contains("server core"),
                "partOfDomain": system["partOfDomain"] == true,
                "domain": clean_text(&js_string(&system["domain"]), 120),
                "powershell": format!("{} {}", js_string(&system["psVersion"]), js_string(&system["psEdition"])),
                "is64BitProcess": system["is64BitProcess"] == true,
                "defaultShell": platform["defaultShell"].as_str().unwrap_or("cmd.exe"),
                "sshLayout": match &managed { Some(m) if m.analysis.supported => "ok", Some(_) => "unsupported", None => "unreadable" },
                "sshVersion": clean_text(&js_string(&raw["ssh"]["version"]), 120),
                "sshService": clean_text(&js_string(&raw["ssh"]["service"]), 40),
                "managedPath": winpolicy::CONFIG_PATH,
                "tools": []
            },
            "shells": [],
            "groups": groups.iter().map(|g| json!({ "name": g["name"], "sid": g["sid"], "description": g["description"], "members": g["members"].as_array().map(|m| m.iter().map(|m| m["sid"].clone()).collect::<Vec<_>>()).unwrap_or_default() })).collect::<Vec<_>>(),
            "adminGroups": ["Administrators"],
            "externalIdentities": external,
            "managedPolicy": {
                "state": match &managed { Some(m) if !m.policies.is_empty() => "present", Some(_) => "absent", None => "unreadable" },
                "consistent": managed.as_ref().is_some_and(|m| m.consistent),
                "users": managed.as_ref().map(|m| m.policies.keys().cloned().collect::<Vec<_>>()).unwrap_or_default()
            },
            "users": users,
            "collectedAt": now_iso()
        })
    }

    fn resolve_target(payload: &Value) -> CoreResult<Value> {
        if truthy(&payload["sid"]) {
            return Ok(json!({ "sid": validate_sid(&payload["sid"], &LOCAL_SID, "USERS_SID_INVALID")? }));
        }
        let name = if payload.get("name").is_some_and(|n| !n.is_null()) { &payload["name"] } else { &payload["username"] };
        Ok(json!({ "name": validate_windows_name(name)? }))
    }

    pub async fn detail(&self, session_id: &str, payload: &Value, platform: &Value) -> CoreResult<WinDetail> {
        let target = Self::resolve_target(payload)?;
        let privilege = self.privileges(session_id).await?;
        let addresses = family_addresses(privilege["clientAddress"].as_str().unwrap_or(""));
        let mut args = json!({ "op": "detail", "addr4": addresses.ipv4, "addr6": addresses.ipv6 });
        for (k, v) in target.as_object().unwrap() {
            args[k] = v.clone();
        }
        let raw = self.run(session_id, args, "users.windows.detail", 120_000).await?;
        self.map_detail(session_id, &raw, privilege, addresses, platform)
    }

    fn map_detail(&self, session_id: &str, raw: &Value, privilege: Value, addresses: Addresses, platform: &Value) -> CoreResult<WinDetail> {
        let session = self.sessions.session_auth_info(session_id)?;
        let user = &raw["user"];
        let sid = js_string(&user["sid"]);
        let rid: i64 = sid.rsplit('-').next().and_then(|r| r.parse().ok()).unwrap_or(0);
        let groups: Vec<Value> = raw["groups"].as_array().cloned().unwrap_or_default().into_iter().map(|g| json!({ "sid": g["sid"], "name": js_string(&g["name"]) })).collect();
        let admin = groups.iter().any(|g| g["sid"] == ADMINS_SID);
        let name = winpolicy::ssh_name(&js_string(&user["name"]));
        let config = decode_config(&raw["ssh"]);
        let managed = (raw["ssh"]["configReadable"] == true).then(|| winpolicy::parse_managed_policies(&config));
        let policy = match (&name, &managed) {
            (Some(n), Some(m)) => m.policies.get(n).cloned().unwrap_or_default(),
            _ => Policy::default(),
        };
        let well_known = |sid: &str| match sid {
            "S-1-5-32-544" => Some("administrators"),
            "S-1-5-32-545" => Some("users"),
            "S-1-5-32-585" => Some("openssh users"),
            _ => None,
        };
        let mut group_names: Vec<String> = Vec::new();
        for g in &groups {
            for candidate in [Some(js_string(&g["name"]).to_lowercase()), well_known(g["sid"].as_str().unwrap_or("")).map(str::to_string)].into_iter().flatten() {
                if !candidate.is_empty() && !group_names.contains(&candidate) {
                    group_names.push(candidate);
                }
            }
        }
        let mut effective = Map::new();
        let mut warnings: Vec<String> = Vec::new();
        let mut listener: Option<Value> = None;
        for (family, key) in [("ipv4", "effective4"), ("ipv6", "effective6")] {
            if raw[key].is_null() || name.is_none() {
                continue;
            }
            let entry = &raw[key];
            let mut parsed = policy::parse_effective(if entry["code"] == 0 { entry["text"].as_str().unwrap_or("") } else { "" });
            if entry["code"] != 0 {
                parsed.errors.push(clean_text(&js_string(&entry["text"]), 300));
            }
            let address = addresses.get(family);
            let evaluation = policy::evaluate_access(&parsed, &AccessContext { username: name.as_deref().unwrap_or(""), groups: &group_names, address: &address, method: None, keyboard_interactive: false });
            effective.insert(
                family.into(),
                json!({
                    "address": address,
                    "sample": addresses.sample(family),
                    "passwordAuthentication": parsed.get("passwordauthentication").unwrap_or("unknown"),
                    "pubkeyAuthentication": parsed.get("pubkeyauthentication").unwrap_or("unknown"),
                    "authenticationMethods": parsed.get("authenticationmethods").unwrap_or("any"),
                    "allowed": evaluation.allowed_value(),
                    "reasons": evaluation.reasons,
                    "methods": evaluation.methods,
                    "errors": parsed.errors.iter().take(5).collect::<Vec<_>>()
                }),
            );
            if listener.is_none() {
                listener = Some(policy::listener_families(&parsed));
            }
        }
        if let Some(l) = &listener {
            effective.insert("listener".into(), l.clone());
            if l["ipv6"] == false {
                warnings.push("sshd does not listen on IPv6; IPv6 logins are impossible regardless of the account policy.".into());
            }
            if l["ipv4"] == false {
                warnings.push("sshd does not listen on IPv4; IPv4 logins are impossible regardless of the account policy.".into());
            }
        }
        if name.is_none() {
            warnings.push("The account name contains characters that cannot be used safely in sshd_config; CH-J does not manage its SSH policy.".into());
        }
        if let Some(m) = &managed {
            if !m.analysis.supported {
                warnings.push(format!("sshd_config cannot be edited safely: {}.", m.analysis.problems.join("; ")));
            } else if !m.consistent {
                warnings.push("The CH-J sections in sshd_config were edited manually; CH-J will not overwrite them.".into());
            }
        }
        let profile_path = raw["profilePath"].as_str().filter(|s| !s.is_empty()).map(str::to_string);
        let key_files: Vec<WinKeyFile> = raw["keyFiles"].as_array().cloned().unwrap_or_default().iter().map(|f| map_key_file(f, &sid, profile_path.as_deref())).collect();
        if key_files.iter().any(|f| f.shared) {
            warnings.push("Administrators share one administrators_authorized_keys file: every key in it can log in as any member of the Administrators group.".into());
        }
        for file in &key_files {
            warnings.extend(file.acl_warnings.iter().cloned());
        }
        let self_account = sid == privilege["sid"].as_str().unwrap_or("");
        let ssh = &raw["ssh"];
        let public = json!({
            "platform": "windows",
            "sessionId": session_id,
            "username": user["name"],
            "sid": sid,
            "privileged": privilege["elevated"],
            "self": self_account,
            "session": { "authMethod": session["authMethod"], "username": privilege["username"], "sid": privilege["sid"], "clientAddress": privilege["clientAddress"] },
            "privilege": privilege,
            "account": {
                "platform": "windows",
                "username": user["name"],
                "sid": sid,
                "rid": rid,
                "enabled": user["enabled"] == true,
                "fullName": clean_text(&js_string(&user["fullName"]), 256),
                "comment": clean_text(&js_string(&user["fullName"]), 256),
                "description": clean_text(&js_string(&user["description"]), 256),
                "source": user["source"],
                "lastLogon": user["lastLogon"],
                "passwordLastSet": user["passwordLastSet"],
                "passwordRequired": user["passwordRequired"] == true,
                "passwordExpires": user["passwordExpires"],
                "accountExpires": user["accountExpires"],
                "profilePath": profile_path,
                "groups": groups,
                "admin": admin,
                "system": rid < 1000,
                "protection": if rid < 1000 { json!("builtin") } else if self_account { json!("session") } else { Value::Null }
            },
            "ssh": {
                "available": ssh["installed"] == true,
                "version": clean_text(&js_string(&ssh["version"]), 160),
                "service": format!("{} / {}", ssh["service"].as_str().unwrap_or("unknown"), ssh["startType"].as_str().unwrap_or("unknown")),
                "configPath": ssh["configPath"],
                "layout": match &managed { Some(m) if m.analysis.supported => "ok", Some(_) => "unsupported", None => "unreadable" },
                "layoutSupported": managed.as_ref().is_some_and(|m| m.analysis.supported),
                "layoutProblems": managed.as_ref().map(|m| m.analysis.problems.clone()).unwrap_or_else(|| vec!["sshd_config is not readable".into()]),
                "syntaxOk": if raw["syntax"].is_object() { json!(raw["syntax"]["code"] == 0) } else { Value::Null },
                "syntaxMessage": clean_text(&js_string(&raw["syntax"]["output"]), 400),
                "managed": {
                    "state": match &managed { Some(m) if !m.policies.is_empty() => "present", Some(_) => "absent", None => "unreadable" },
                    "consistent": managed.as_ref().is_some_and(|m| m.consistent),
                    "path": winpolicy::CONFIG_PATH
                },
                "restartRequired": true,
                "sshName": name,
                "policy": policy.to_value(),
                "effective": Value::Object(effective),
                "warnings": warnings
            },
            "keys": { "files": key_files.iter().map(|f| f.public.clone()).collect::<Vec<_>>(), "home": profile_path },
            "system": if platform.is_null() { Value::Null } else { platform.clone() },
            "collectedAt": now_iso()
        });
        Ok(WinDetail {
            public,
            key_files,
            config_text: config,
            config_sha256: ssh["configSha256"].clone(),
            managed,
            addresses,
            groups: group_names,
            privilege,
            sid,
            username: js_string(&user["name"]),
            self_account,
        })
    }

    fn assert_mutable(account: &Value) -> CoreResult<()> {
        match account["protection"].as_str() {
            Some("builtin") => return Err(fail("USERS_PROTECTED_USER", "Built-in Windows accounts (Administrator, Guest, DefaultAccount, WDAGUtilityAccount) are protected.")),
            Some("session") => return Err(fail("USERS_PROTECTED_USER", "The account used by the active SSH session is protected against changes that could lock you out.")),
            _ => {}
        }
        let source = js_string(&account["source"]);
        if !source.is_empty() && !["local", "microsoftaccount"].contains(&source.to_lowercase().as_str()) {
            return Err(fail("USERS_DOMAIN_READ_ONLY", "Domain and cloud identities are shown read-only; CH-J does not manage them."));
        }
        Ok(())
    }

    fn require_elevated(privilege: &Value) -> CoreResult<()> {
        if privilege["elevated"] == true {
            return Ok(());
        }
        Err(fail(
            "USERS_WINDOWS_ELEVATION_REQUIRED",
            if privilege["adminMember"] == true {
                "The SSH session belongs to Administrators, but its token is not elevated (UAC filtered). CH-J does not bypass UAC."
            } else {
                "The SSH account is not a local administrator; administrative operations are not available."
            },
        ))
    }

    pub async fn manage(&self, session_id: &str, payload: &Value, platform: &Value) -> CoreResult<Value> {
        let privilege = self.privileges(session_id).await?;
        Self::require_elevated(&privilege)?;
        let requested = js_string(&payload["action"]);
        let result = if requested == "create" {
            let name = validate_windows_name(&payload["username"])?;
            let mut groups: Vec<String> = Vec::new();
            for sid in payload["groups"].as_array().cloned().unwrap_or_default() {
                let sid = validate_sid(&sid, &GROUP_SID, "USERS_GROUP_INVALID")?;
                if !groups.contains(&sid) {
                    groups.push(sid);
                }
            }
            let full_name = if payload.get("fullName").is_some_and(|v| !v.is_null()) { &payload["fullName"] } else { &payload["comment"] };
            let mut args = json!({
                "op": "create",
                "name": name,
                "fullName": validate_text(full_name, 256, "USERS_COMMENT_INVALID")?,
                "description": validate_text(&payload["description"], 48, "USERS_COMMENT_INVALID")?,
                "admin": payload["admin"] == true,
                "groups": groups
            });
            if truthy(&payload["password"]) {
                args["password"] = json!(validate_password(&payload["password"])?);
            }
            self.run(session_id, args, "users.windows.create", 120_000).await?
        } else {
            let sid = validate_sid(&payload["sid"], &LOCAL_SID, "USERS_SID_INVALID")?;
            let inventory = self.inventory(session_id, platform).await?;
            let account = inventory["users"].as_array().and_then(|u| u.iter().find(|u| u["sid"] == sid.as_str())).cloned().ok_or_else(|| fail("USERS_USER_NOT_FOUND", "The local account does not exist."))?;
            Self::assert_mutable(&account)?;
            if requested == "delete" {
                self.run(session_id, json!({ "op": "delete", "sid": sid, "removeProfile": payload["removeProfile"] == true || payload["removeHome"] == true }), "users.windows.delete", 120_000).await?
            } else {
                let action = action_name(&requested).ok_or_else(|| fail("USERS_NOT_SUPPORTED_ON_PLATFORM", "This operation exists only on Linux (shell, home directory, primary group, password lock)."))?;
                let mut args = json!({ "op": "account", "action": action, "sid": sid });
                match action {
                    "password" => args["password"] = json!(validate_password(&payload["password"])?),
                    "fullName" => {
                        let value = if payload.get("fullName").is_some_and(|v| !v.is_null()) { &payload["fullName"] } else { &payload["comment"] };
                        args["value"] = json!(validate_text(value, 256, "USERS_COMMENT_INVALID")?);
                    }
                    "description" => args["value"] = json!(validate_text(&payload["description"], 48, "USERS_COMMENT_INVALID")?),
                    "group-add" | "group-remove" => args["group"] = json!(validate_sid(&payload["group"], &GROUP_SID, "USERS_GROUP_INVALID")?),
                    _ => {}
                }
                self.run(session_id, args, &format!("users.windows.{action}"), 120_000).await?
            }
        };
        self.logger.warn("Remote Windows account action completed.", json!({ "sessionId": session_id, "action": requested, "result": result["result"] }));
        let mut warnings: Vec<String> = Vec::new();
        if requested == "delete" && truthy(&payload["sid"]) {
            if let Err(error) = self.cleanup_policy(session_id, payload, &mut warnings).await {
                warnings.push(format!("CH-J SSH rules of the deleted account remain: {}", error.message));
            }
        }
        let inventory = self.inventory(session_id, platform).await?;
        let outcome = if !warnings.is_empty() && result["result"] == "ok" { json!("partial") } else { result["result"].clone() };
        Ok(json!({
            "action": requested,
            "username": js_string(&payload["username"]),
            "sid": if truthy(&result["sid"]) { result["sid"].clone() } else if truthy(&payload["sid"]) { payload["sid"].clone() } else { Value::Null },
            "result": outcome,
            "steps": result["steps"].as_array().cloned().unwrap_or_default(),
            "warnings": warnings,
            "users": inventory["users"],
            "inventory": inventory
        }))
    }

    async fn cleanup_policy(&self, session_id: &str, payload: &Value, warnings: &mut Vec<String>) -> CoreResult<()> {
        let Some(name) = winpolicy::ssh_name(&js_string(&payload["username"])) else { return Ok(()) };
        let session = self.privileges(session_id).await?;
        let context = self.detail(session_id, &json!({ "sid": session["sid"] }), &Value::Null).await?;
        if !context.managed.as_ref().is_some_and(|m| m.policies.contains_key(&name)) {
            return Ok(());
        }
        let credentials = payload.get("probeCredentials").filter(|v| v.is_object()).cloned().unwrap_or_else(|| json!({}));
        let result = self.apply_policy(session_id, &context, &name, None, &credentials, false).await?;
        warnings.extend(result["warnings"].as_array().cloned().unwrap_or_default().into_iter().filter_map(|w| w.as_str().map(str::to_string)));
        Ok(())
    }

    pub async fn keys(&self, session_id: &str, payload: &Value, platform: &Value) -> CoreResult<Value> {
        let action = js_string(&payload["action"]);
        if action != "add" && action != "remove" {
            return Err(fail("USERS_ACTION_INVALID", "Unsupported key action."));
        }
        let target = Self::resolve_target(payload)?;
        let detail = self.detail(session_id, &target, platform).await?;
        let session = self.sessions.session_auth_info(session_id)?;
        if detail.public["account"]["protection"] == "builtin" {
            return Err(fail("USERS_PROTECTED_USER", "Keys of built-in accounts are shown read-only."));
        }
        if !detail.self_account {
            Self::require_elevated(&detail.privilege)?;
        }
        let (file, content) = if action == "add" {
            let key = keys::normalize_public_key(&payload["publicKey"], payload.get("comment").unwrap_or(&Value::Null))?;
            for candidate in &detail.key_files {
                if candidate.entries.iter().any(|e| e.blob.as_deref() == Some(key.blob.as_str())) {
                    return Err(fail("USERS_KEY_DUPLICATE", &format!("This public key is already authorized in {}.", candidate.path)));
                }
            }
            let file = detail.key_files.first().filter(|f| f.manageable).ok_or_else(|| {
                let reason = detail.key_files.first().and_then(|f| f.public["reason"].as_str().map(str::to_string)).unwrap_or_else(|| "not found".into());
                fail("USERS_KEYS_FILE_UNMANAGEABLE", &format!("The effective authorized_keys file cannot be managed safely ({reason})."))
            })?;
            (file, keys::append_key(&file.raw, &key)?)
        } else {
            let file = detail.key_files.iter().find(|f| Some(f.path.as_str()) == payload["path"].as_str()).filter(|f| f.manageable).ok_or_else(|| fail("USERS_KEYS_FILE_UNMANAGEABLE", "This key file cannot be changed safely."))?;
            let fingerprint = js_string(&payload["fingerprint"]);
            let session_uses_file = detail.self_account || (file.shared && detail.privilege["adminMember"] == true);
            if session_uses_file && session["authMethod"] == "privateKey" {
                if session["publicKeyFingerprint"].as_str().is_some_and(|fp| fp == fingerprint) {
                    return Err(fail("USERS_KEY_IN_USE", "This key is used by the active CH-J session and is not removed."));
                }
                if file.entries.iter().filter(|e| e.valid).count() <= 1 {
                    return Err(fail("USERS_KEY_LAST", "The last key of the file used by the active session is not removed."));
                }
            }
            (file, keys::remove_key(&file.raw, &payload["line"], &fingerprint)?)
        };
        let expected = if file.kind == "file" { keys::sha256_hex(&file.raw) } else { "absent".into() };
        self.run(
            session_id,
            json!({ "op": "keys-write", "sid": detail.sid, "path": file.path, "expected": expected, "data": B64.encode(&content), "addr4": detail.addresses.ipv4, "addr6": detail.addresses.ipv6 }),
            &format!("users.windows.keys.{action}"),
            120_000,
        )
        .await?;
        self.logger.warn("Remote Windows authorized keys changed.", json!({ "sessionId": session_id, "action": action, "path": file.path, "shared": file.shared }));
        let (path, shared) = (file.path.clone(), file.shared);
        let refreshed = self.detail(session_id, &json!({ "sid": detail.sid }), platform).await?;
        Ok(json!({ "action": action, "username": detail.username, "sid": detail.sid, "path": path, "shared": shared, "detail": refreshed.public }))
    }

    pub async fn ssh_policy(&self, session_id: &str, payload: &Value, platform: &Value) -> CoreResult<Value> {
        let target = Self::resolve_target(payload)?;
        let policy = policy::normalize_policy(payload.get("policy").filter(|p| p.is_object()).unwrap_or(&json!({})), Some("windows-account"))?;
        let detail = self.detail(session_id, &target, platform).await?;
        Self::require_elevated(&detail.privilege)?;
        if detail.public["account"]["protection"] == "builtin" {
            return Err(fail("USERS_PROTECTED_USER", "SSH access of built-in accounts is not managed by CH-J."));
        }
        let name = detail.public["ssh"]["sshName"].as_str().map(str::to_string).ok_or_else(|| fail("USERS_SSH_NAME_UNSUPPORTED", "The account name cannot be used safely in sshd_config."))?;
        let credentials = payload.get("probeCredentials").filter(|v| v.is_object()).cloned().unwrap_or_else(|| json!({}));
        let mut result = self.apply_policy(session_id, &detail, &name, Some(policy), &credentials, detail.self_account).await?;
        match self.detail(session_id, &json!({ "sid": detail.sid }), platform).await {
            Ok(refreshed) => result["detail"] = refreshed.public,
            Err(error) if error.code == "SESSION_NOT_CONNECTED" => {
                result["sessionLost"] = json!(true);
                result["detail"] = Value::Null;
            }
            Err(error) => return Err(error),
        }
        Ok(result)
    }

    /// Stage → commit through a SYSTEM scheduled task that arms the rollback before restarting
    /// sshd → confirm from a NEW SSH connection → otherwise the task restores the backup.
    async fn apply_policy(&self, session_id: &str, detail: &WinDetail, name: &str, policy: Option<Policy>, credentials: &Value, self_account: bool) -> CoreResult<Value> {
        let privilege = &detail.privilege;
        let ssh = &detail.public["ssh"];
        if ssh["available"] != true {
            return Err(fail("USERS_SSHD_MISSING", "OpenSSH Server (sshd.exe) is not installed."));
        }
        if ssh["syntaxOk"] == false {
            return Err(fail("USERS_SSH_CONFIG_ALREADY_INVALID", &format!("The current sshd_config is invalid: {}", ssh["syntaxMessage"].as_str().unwrap_or(""))));
        }
        let managed = detail.managed.as_ref().ok_or_else(|| fail("USERS_SSH_LAYOUT_UNSUPPORTED", "sshd_config is not readable."))?;
        let mut policies: Policies = managed.policies.clone();
        match &policy {
            Some(p) if !p.is_default() => {
                policies.insert(name.to_string(), p.clone());
            }
            _ => {
                policies.remove(name);
            }
        }
        let content = winpolicy::render_config(&detail.config_text, &policies)?;
        let result_policy = policy.clone().unwrap_or_default().to_value();
        if content == detail.config_text {
            return Ok(json!({ "changed": false, "applied": false, "policy": result_policy, "warnings": [] }));
        }
        let client_address = privilege["clientAddress"].as_str().unwrap_or("").to_string();
        if client_address.parse::<std::net::IpAddr>().is_err() {
            return Err(fail("USERS_SSH_CLIENT_UNKNOWN", "The client address of the SSH session is unknown, so lockout protection cannot be verified."));
        }
        // Every Windows change restarts sshd, so a fresh login is always required to keep it.
        let requirements = self.sessions.probe_login_requirements(session_id, credentials)?;
        if requirements["ready"] != true {
            return Err(fail("USERS_SSH_PROBE_CREDENTIALS_REQUIRED", &format!("Windows OpenSSH must be restarted; a test login ({}) is required to keep the change.", requirements["missing"].as_str().unwrap_or(""))));
        }
        let target = self.sessions.connection_target(session_id)?;
        let session = self.sessions.session_auth_info(session_id)?;
        let token_name = account_name(privilege["username"].as_str().unwrap_or(""));
        let session_name = winpolicy::ssh_name(&token_name).unwrap_or_else(|| token_name.to_lowercase());
        let method = if session["authMethod"] == "privateKey" { "publickey" } else { "password" };
        let addresses = &detail.addresses;
        let mut tuples: Vec<Value> = Vec::new();
        if policy.is_some() {
            tuples.push(json!({ "user": name, "addr": addresses.ipv4 }));
            tuples.push(json!({ "user": name, "addr": addresses.ipv6 }));
        }
        tuples.push(json!({ "user": session_name, "addr": client_address }));
        let staged = self
            .run(session_id, json!({ "op": "ssh-stage", "expected": detail.config_sha256, "data": B64.encode(content.as_bytes()), "tuples": tuples }), "users.windows.ssh-stage", 120_000)
            .await?;
        if staged["syntaxCode"] != 0 {
            let output = clean_text(&js_string(&staged["syntaxOutput"]), 400);
            return Err(fail("USERS_SSH_CONFIG_INVALID", if output.is_empty() { "sshd -t rejected the new configuration." } else { &output }));
        }
        let parse = |entry: &Value| -> Effective {
            let mut parsed = policy::parse_effective(if entry["code"] == 0 { entry["text"].as_str().unwrap_or("") } else { "" });
            if !entry.is_object() || entry["code"] != 0 {
                let text = js_string(&entry["text"]);
                parsed.errors.push(clean_text(if text.is_empty() { "missing" } else { &text }, 200));
            }
            parsed
        };
        let mut problems: Vec<String> = Vec::new();
        if let Some(policy) = &policy {
            for (family, label) in [("ipv4", "IPv4"), ("ipv6", "IPv6")] {
                let effective = parse(&staged["after"][format!("{name}|{}", addresses.get(family))]);
                if !effective.errors.is_empty() {
                    problems.push(format!("{label}: sshd -T did not return a usable result"));
                } else {
                    problems.extend(policy::verify_policy_effect(policy, name, &effective, label, &VerifyOptions { keyboard_interactive: false, ipv4_pattern: policy::WINDOWS_IPV4_PATTERN, ipv6_pattern: policy::WINDOWS_IPV6_PATTERN }));
                }
                // With UseDNS the host part is also matched against reverse-DNS names, where *.*.*.* is not IPv4-specific.
                if (policy.ipv4 == "deny" || policy.ipv6 == "deny") && effective.get("usedns") == Some("yes") {
                    problems.push(format!("{label}: UseDNS yes makes address-family rules ambiguous"));
                }
            }
        }
        if !problems.is_empty() {
            return Err(fail("USERS_SSH_POLICY_NOT_EFFECTIVE", &format!("Existing sshd rules take precedence ({}). Nothing was changed.", problems.join("; "))));
        }
        let session_key = format!("{session_name}|{client_address}");
        let before = parse(&staged["before"][&session_key]);
        let after = parse(&staged["after"][&session_key]);
        let groups: Vec<String> = if detail.self_account { detail.groups.clone() } else { privilege["groups"].as_array().map(|g| g.iter().filter_map(|v| v.as_str().map(str::to_string)).collect()).unwrap_or_default() };
        let context = AccessContext { username: &session_name, groups: &groups, address: &client_address, method: Some(method), keyboard_interactive: false };
        let evaluation_after = policy::evaluate_access(&after, &context);
        if !after.errors.is_empty() || evaluation_after.method_allowed == Some(false) || evaluation_after.allowed == Some(false) {
            return Err(fail("USERS_SSH_LOCKOUT_PREVENTED", &format!("The active session ({method} from {client_address}) would no longer be allowed to log in. Nothing was changed.")));
        }
        if !self_account {
            let comparable = |effective: &Effective| {
                let evaluation = policy::evaluate_access(effective, &context);
                let settings: Vec<(&String, &Value)> = effective.values.iter().filter(|(k, _)| !["denyusers", "allowusers", "denygroups", "allowgroups"].contains(&k.as_str())).collect();
                serde_json::to_string(&json!({ "evaluation": { "allowed": evaluation.allowed_value(), "reasons": evaluation.reasons, "methods": evaluation.methods, "methodAllowed": evaluation.method_allowed }, "settings": settings })).unwrap_or_default()
            };
            if comparable(&before) != comparable(&after) {
                return Err(fail("USERS_SSH_LOCKOUT_PREVENTED", "The change would alter the effective SSH access of the active session account. Nothing was changed."));
            }
        }
        let mut nonce_bytes = [0u8; 24];
        rand::rngs::OsRng.fill_bytes(&mut nonce_bytes);
        let nonce = hex::encode(nonce_bytes);
        let commit = self
            .run(
                session_id,
                json!({ "op": "ssh-commit", "id": staged["id"], "nonce": nonce, "expected": detail.config_sha256, "candidateSha256": staged["candidateSha256"], "timeoutSeconds": APPLY_TIMEOUT_SECONDS }),
                "users.windows.ssh-commit",
                120_000,
            )
            .await?;
        self.logger.warn("Windows sshd change armed with rollback guard.", json!({ "sessionId": session_id, "id": staged["id"], "state": commit["state"] }));
        if commit["state"] != "awaiting-confirmation" {
            return Err(fail("USERS_SSH_ROLLED_BACK", &format!("The change was not kept ({}): {}", js_string(&commit["state"]), clean_text(&js_string(&commit["message"]), 300))));
        }
        let confirm_args = |confirm: bool, wait: u64| json!({ "op": "ssh-confirm", "id": staged["id"], "nonce": nonce, "confirm": confirm, "waitSeconds": wait });
        let confirmation = match self.sessions.fresh_exec(&target, credentials, &encoded_command(BOOTSTRAP), &bootstrap_input(REMOTE_SCRIPT, &confirm_args(true, 25)), Duration::from_secs(90), true).await {
            Ok(fresh) => parse_windows_output(fresh["stdout"].as_str().unwrap_or(""), fresh["stderr"].as_str().unwrap_or(""), fresh["code"].as_i64(), "USERS_")?,
            Err(error) => {
                let rollback = self.await_rollback(session_id, &target, credentials, confirm_args(false, APPLY_TIMEOUT_SECONDS + 30)).await;
                return Err(fail("USERS_SSH_PROBE_FAILED", &format!("{} The rollback guard reported: {rollback}", error.message)));
            }
        };
        if confirmation["state"] != "confirmed" {
            return Err(fail("USERS_SSH_ROLLED_BACK", &format!("The change was not kept ({}): {}", js_string(&confirmation["state"]), clean_text(&js_string(&confirmation["message"]), 300))));
        }
        self.logger.warn("Windows sshd change confirmed by a new SSH login.", json!({ "sessionId": session_id, "id": staged["id"] }));
        let mut warnings: Vec<String> = Vec::new();
        if self.sessions.require_connected(session_id).is_err() {
            warnings.push("The original SSH session was closed by the sshd restart. Reconnect in the main window.".into());
        }
        Ok(json!({ "changed": true, "applied": true, "probed": true, "restarted": true, "policy": result_policy, "applyId": staged["id"], "warnings": warnings }))
    }

    /// Waits for the server-side guard to restore the configuration, through the old session if it is still alive.
    async fn await_rollback(&self, session_id: &str, target: &crate::core::ssh::session::ConnectionTarget, credentials: &Value, args: Value) -> String {
        if self.sessions.require_connected(session_id).is_ok() {
            if let Ok(status) = self.executor.run(session_id, REMOTE_SCRIPT, &args, Duration::from_secs(APPLY_TIMEOUT_SECONDS + 60), "users.windows.ssh-confirm", "USERS_").await {
                return format!("{} ({})", js_string(&status["state"]), clean_text(&js_string(&status["message"]), 200));
            }
        }
        tokio::time::sleep(Duration::from_secs(APPLY_TIMEOUT_SECONDS + 15)).await;
        match self.sessions.fresh_exec(target, credentials, &encoded_command(BOOTSTRAP), &bootstrap_input(REMOTE_SCRIPT, &args), Duration::from_secs(APPLY_TIMEOUT_SECONDS + 60), true).await {
            Ok(fresh) => match parse_windows_output(fresh["stdout"].as_str().unwrap_or(""), fresh["stderr"].as_str().unwrap_or(""), fresh["code"].as_i64(), "USERS_") {
                Ok(status) => format!("{} ({})", js_string(&status["state"]), clean_text(&js_string(&status["message"]), 200)),
                Err(error) => format!("unknown ({}); the scheduled task restores the previous sshd_config automatically after {APPLY_TIMEOUT_SECONDS} s", clean_text(&error.message, 200)),
            },
            Err(error) => format!("unknown ({}); the scheduled task restores the previous sshd_config automatically after {APPLY_TIMEOUT_SECONDS} s", clean_text(&error.message, 200)),
        }
    }
}

static WRITE_RIGHTS: Lazy<Regex> = Lazy::new(|| Regex::new(r"Write|Modify|FullControl|ChangePermissions|TakeOwnership").unwrap());

fn map_key_file(file: &Value, owner_sid: &str, profile_path: Option<&str>) -> WinKeyFile {
    let raw = file["data"].as_str().filter(|s| !s.is_empty()).map(|d| B64.decode(d).unwrap_or_default()).unwrap_or_default();
    let kind = js_string(&file["type"]);
    let shared = file["shared"] == true;
    let path = js_string(&file["path"]);
    let reason: Option<&str> = if path.is_empty() || kind == "unresolved" {
        Some("unsupported-token")
    } else if kind == "needs-elevation" {
        Some("needs-elevation")
    } else if file["reparse"] == true || kind == "reparse" {
        Some("symlink")
    } else if kind == "other" {
        Some("not-regular")
    } else if kind == "unreadable" {
        Some("unreadable")
    } else if file["tooLarge"] == true {
        Some("too-large")
    } else if kind == "absent-dir" && (shared || profile_path.is_none()) {
        Some("profile-missing")
    } else if !["file", "absent", "absent-dir"].contains(&kind.as_str()) {
        Some("unknown")
    } else {
        None
    };
    let mut allowed = vec!["S-1-5-18".to_string(), ADMINS_SID.to_string()];
    if !shared {
        allowed.push(owner_sid.to_string());
    }
    let acl: Vec<Value> = file["acl"].as_array().cloned().unwrap_or_default();
    let mut acl_warnings = Vec::new();
    for entry in &acl {
        let sid = js_string(&entry["sid"]);
        if entry["type"] == "Allow" && !allowed.contains(&sid) && WRITE_RIGHTS.is_match(&js_string(&entry["rights"])) {
            acl_warnings.push(format!("{path}: {sid} has {}; Windows OpenSSH ignores key files writable by other principals.", js_string(&entry["rights"])));
        }
    }
    let entries = if kind == "file" { keys::parse_authorized_keys(&raw) } else { Vec::new() };
    let owner = file["owner"].as_str().filter(|s| !s.is_empty()).map(str::to_string);
    let public = json!({
        "path": path,
        "spec": clean_text(&js_string(&file["spec"]), 200),
        "type": kind,
        "shared": shared,
        "owner": owner,
        "acl": acl.iter().map(|e| json!({ "sid": e["sid"], "rights": clean_text(&js_string(&e["rights"]), 120), "type": e["type"], "inherited": e["inherited"] == true })).collect::<Vec<_>>(),
        "meta": owner.as_ref().map(|o| format!("owner {o}")).unwrap_or_default(),
        "dirMeta": "",
        "sha256": file["sha256"].as_str().filter(|s| !s.is_empty()).unwrap_or("absent"),
        "manageable": reason.is_none(),
        "reason": reason,
        "aclWarnings": acl_warnings,
        "entries": entries.iter().map(keys::KeyEntry::public).collect::<Vec<_>>()
    });
    WinKeyFile { path, kind, shared, manageable: reason.is_none(), entries, raw, public, acl_warnings }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn token_and_names() {
        let privilege = privilege_from_token(&json!({ "name": "HOST\\Codex", "sid": "S-1-5-21-1-2-3-1001", "elevated": false, "adminMember": true, "groupNames": ["Administrators"] }), &json!({ "sshClient": "192.0.2.5 50000 22" }));
        assert_eq!(privilege["mode"], "windows-admin-unelevated");
        assert_eq!(privilege["clientAddress"], "192.0.2.5");
        assert_eq!(privilege["groups"][0], "administrators");
        assert_eq!(account_name("HOST\\codex"), "codex");
        assert!(validate_windows_name(&json!("Bad.")).is_err());
        assert!(validate_sid(&json!("s-1-5-21-1-2-3-1001"), &LOCAL_SID, "X").is_ok());
        assert!(validate_sid(&json!("S-1-5-32-544"), &LOCAL_SID, "X").is_err());
        let file = map_key_file(&json!({ "path": "C:\\Users\\a\\.ssh\\authorized_keys", "type": "file", "data": "", "acl": [{ "sid": "S-1-1-0", "rights": "Modify", "type": "Allow" }] }), "S-1-5-21-1-2-3-1001", Some("C:\\Users\\a"));
        assert!(file.manageable);
        assert_eq!(file.acl_warnings.len(), 1);
    }
}