Nexus/server/utils/files_upload_permissions.py

"""Post-upload permission fixup for file manager (reuse rsync push defaults)."""

from __future__ import annotations

import logging
import re
import shlex
from typing import TYPE_CHECKING

from server.application.services.sync_engine_v2 import _RSYNC_CHMOD_RE, _RSYNC_CHOWN_RE

if TYPE_CHECKING:
    from server.domain.models import Server

logger = logging.getLogger(__name__)


def parse_upload_dir_mode(chmod_setting: str) -> str | None:
    """Extract octal directory mode from rsync-style ``D755,F755``."""
    raw = (chmod_setting or "").strip()
    if not raw or not _RSYNC_CHMOD_RE.fullmatch(raw):
        return None
    m = re.search(r"D([0-7]{3,4})", raw, re.IGNORECASE)
    if m:
        return m.group(1)
    return parse_upload_file_mode(raw)


def parse_upload_file_mode(chmod_setting: str) -> str | None:
    """Extract octal file mode from rsync-style ``D755,F755`` or plain ``644``."""
    raw = (chmod_setting or "").strip()
    if not raw or not _RSYNC_CHMOD_RE.fullmatch(raw):
        return None
    m = re.search(r"F([0-7]{3,4})", raw, re.IGNORECASE)
    if m:
        return m.group(1)
    if re.fullmatch(r"[0-7]{3,4}", raw):
        return raw
    for part in reversed(raw.split(",")):
        piece = part.strip()
        if len(piece) > 1 and piece[0] in "Ff" and re.fullmatch(r"[0-7]{3,4}", piece[1:]):
            return piece[1:]
    return None


def parse_upload_chown(chown_setting: str) -> tuple[str, str] | None:
    """Return ``(owner, group)`` when setting is valid; else None."""
    raw = (chown_setting or "").strip()
    if not raw or not _RSYNC_CHOWN_RE.fullmatch(raw):
        return None
    if ":" in raw:
        owner, group = raw.split(":", 1)
        return owner.strip(), group.strip()
    return raw, raw


def upload_permission_plan() -> tuple[tuple[str, str] | None, str | None]:
    """Read ``NEXUS_RSYNC_PUSH_CHOWN`` / ``NEXUS_RSYNC_PUSH_CHMOD`` for uploads."""
    from server.config import settings

    return (
        parse_upload_chown(settings.RSYNC_PUSH_CHOWN or ""),
        parse_upload_file_mode(settings.RSYNC_PUSH_CHMOD or ""),
    )


async def apply_upload_file_permissions(server: Server, remote_path: str) -> dict:
    """chown/chmod uploaded file; mirrors rsync push defaults (www:www, F755)."""
    from server.infrastructure.ssh.remote_shell import exec_ssh_command_with_fallback

    chown_pair, file_mode = upload_permission_plan()
    if not chown_pair and not file_mode:
        return {"applied": False, "skipped": True}

    path_q = shlex.quote(remote_path)
    elevated = False
    errors: list[str] = []

    if chown_pair:
        owner_spec = shlex.quote(f"{chown_pair[0]}:{chown_pair[1]}")
        result = await exec_ssh_command_with_fallback(
            server, f"chown {owner_spec} {path_q}", timeout=30,
        )
        elevated = bool(result.get("elevated"))
        if result.get("exit_code") != 0:
            err = ((result.get("stderr") or "") + (result.get("stdout") or "")).strip()[:200]
            errors.append(err or "chown 失败")

    if file_mode and not errors:
        result = await exec_ssh_command_with_fallback(
            server, f"chmod {file_mode} {path_q}", timeout=30,
        )
        elevated = elevated or bool(result.get("elevated"))
        if result.get("exit_code") != 0:
            err = ((result.get("stderr") or "") + (result.get("stdout") or "")).strip()[:200]
            errors.append(err or "chmod 失败")

    if errors:
        logger.warning(
            "Upload permission fixup failed for %s on server %s: %s",
            remote_path,
            getattr(server, "id", "?"),
            "; ".join(errors),
        )

    owner, group = chown_pair if chown_pair else (None, None)
    return {
        "applied": not errors and bool(chown_pair or file_mode),
        "skipped": False,
        "owner": owner,
        "group": group,
        "mode": file_mode,
        "elevated": elevated,
        "error": "; ".join(errors) if errors else None,
    }


async def apply_transfer_permissions(
    server: Server,
    remote_path: str,
    *,
    recursive: bool = True,
) -> dict:
    """chown/chmod after cross-server relay or package deliver (default www:www, D755/F755)."""
    from server.config import settings
    from server.infrastructure.ssh.remote_shell import exec_ssh_command_with_fallback

    chown_pair, file_mode = upload_permission_plan()
    dir_mode = parse_upload_dir_mode(settings.RSYNC_PUSH_CHMOD or "")
    if not chown_pair and not file_mode:
        return {"applied": False, "skipped": True, "path": remote_path}

    path_q = shlex.quote(remote_path)
    elevated = False
    errors: list[str] = []

    if chown_pair:
        owner_spec = shlex.quote(f"{chown_pair[0]}:{chown_pair[1]}")
        chown_flag = "-R " if recursive else ""
        result = await exec_ssh_command_with_fallback(
            server,
            f"chown {chown_flag}{owner_spec} {path_q}",
            timeout=120,
        )
        elevated = bool(result.get("elevated"))
        if result.get("exit_code") != 0:
            err = ((result.get("stderr") or "") + (result.get("stdout") or "")).strip()[:200]
            errors.append(err or "chown 失败")

    if file_mode and not errors:
        if recursive and dir_mode and dir_mode != file_mode:
            for kind, mode in (("d", dir_mode), ("f", file_mode)):
                result = await exec_ssh_command_with_fallback(
                    server,
                    f"find {path_q} -type {kind} -exec chmod {shlex.quote(mode)} {{}} +",
                    timeout=120,
                )
                elevated = elevated or bool(result.get("elevated"))
                if result.get("exit_code") != 0:
                    err = ((result.get("stderr") or "") + (result.get("stdout") or "")).strip()[:200]
                    errors.append(err or f"chmod {kind} 失败")
                    break
        else:
            mode = file_mode
            chmod_flag = "-R " if recursive else ""
            result = await exec_ssh_command_with_fallback(
                server,
                f"chmod {chmod_flag}{shlex.quote(mode)} {path_q}",
                timeout=120,
            )
            elevated = elevated or bool(result.get("elevated"))
            if result.get("exit_code") != 0:
                err = ((result.get("stderr") or "") + (result.get("stdout") or "")).strip()[:200]
                errors.append(err or "chmod 失败")

    if errors:
        logger.warning(
            "Transfer permission fixup failed for %s on server %s: %s",
            remote_path,
            getattr(server, "id", "?"),
            "; ".join(errors),
        )

    owner, group = chown_pair if chown_pair else (None, None)
    return {
        "applied": not errors and bool(chown_pair or file_mode),
        "skipped": False,
        "path": remote_path,
        "owner": owner,
        "group": group,
        "dir_mode": dir_mode,
        "file_mode": file_mode,
        "elevated": elevated,
        "error": "; ".join(errors) if errors else None,
    }