第 11 章：NAS 核心模块分步实现

11.1 模块 1：存储管理系统

核心类设计

"""nas/storage/disk_manager.py — 磁盘管理核心模块

使用 AI 生成的提示词模板：
角色：Linux 系统编程专家
任务：实现磁盘发现、信息获取、挂载管理
约束：Python 3.11+, 系统命令通过 CommandExecutor 封装
"""

import subprocess
import json
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional


@dataclass
class DiskInfo:
    """磁盘信息数据模型"""
    device: str              # /dev/sda
    model: str = ""          # WDC WD40EFRX-68N
    serial: str = ""         # 序列号
    size_bytes: int = 0      # 总字节数
    transport: str = ""      # sata / usb / nvme
    rotational: bool = True  # True=HDD, False=SSD
    partitions: list[dict] = field(default_factory=list)
    smart_status: str = "unknown"  # ok / warning / error / unknown
    smart_attrs: dict = field(default_factory=dict)


class DiskManager:
    """磁盘管理器 — 发现、信息获取、健康监控"""

    @staticmethod
    def list_disks(include_partitions: bool = False) -> list[DiskInfo]:
        """通过 lsblk 发现所有块设备"""
        cmd = ["lsblk", "-J", "-o",
               "NAME,SIZE,MODEL,SERIAL,TRAN,ROTA,MOUNTPOINT,FSTYPE,UUID"]
        if not include_partitions:
            cmd.insert(2, "-d")  # 只显示顶层设备

        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        if result.returncode != 0:
            raise RuntimeError(f"lsblk 执行失败: {result.stderr}")

        data = json.loads(result.stdout)
        disks = []
        for dev in data.get("blockdevices", []):
            disk = DiskInfo(
                device=f"/dev/{dev['name']}",
                model=dev.get("model", ""),
                serial=dev.get("serial", ""),
                size_bytes=dev.get("size", 0),
                transport=dev.get("tran", ""),
                rotational=dev.get("rota", True),
                partitions=dev.get("children", []),
            )
            # 获取 SMART 状态
            disk.smart_status = DiskManager._get_smart_status(disk.device)
            disks.append(disk)
        return disks

    @staticmethod
    def get_disk_detail(device: str) -> Optional[DiskInfo]:
        """获取单个磁盘的详细信息"""
        disks = DiskManager.list_disks()
        for d in disks:
            if d.device == device:
                d.smart_attrs = DiskManager._get_smart_attributes(device)
                return d
        return None

    @staticmethod
    def _get_smart_status(device: str) -> str:
        """获取 SMART 健康状态"""
        try:
            result = subprocess.run(
                ["smartctl", "-H", device],
                capture_output=True, text=True, timeout=30
            )
            if "PASSED" in result.stdout:
                return "ok"
            elif "FAILED" in result.stdout:
                return "error"
            return "unknown"
        except (subprocess.TimeoutExpired, FileNotFoundError):
            return "unknown"

    @staticmethod
    def _get_smart_attributes(device: str) -> dict:
        """获取详细 SMART 属性"""
        try:
            result = subprocess.run(
                ["smartctl", "-A", device],
                capture_output=True, text=True, timeout=30
            )
            attrs = {}
            for line in result.stdout.split("\n"):
                parts = line.strip().split()
                if len(parts) >= 10 and parts[0].isdigit():
                    attrs[parts[1]] = {
                        "id": int(parts[0]),
                        "value": int(parts[3]),
                        "worst": int(parts[4]),
                        "threshold": int(parts[5]),
                        "raw_value": parts[9],
                    }
            return attrs
        except Exception:
            return {}


class MountManager:
    """挂载管理器 — 挂载、卸载、fstab 管理"""

    MOUNT_BASE = Path("/mnt/nas")

    @classmethod
    def mount(cls, device: str, mount_point: str, fs_type: str = "auto",
              options: list[str] = None) -> bool:
        """挂载设备到指定目录"""
        target = cls.MOUNT_BASE / mount_point.lstrip("/")
        target.mkdir(parents=True, exist_ok=True)

        cmd = ["mount"]
        if fs_type != "auto":
            cmd.extend(["-t", fs_type])
        if options:
            cmd.extend(["-o", ",".join(options)])
        cmd.extend([device, str(target)])

        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if result.returncode != 0:
            raise OSError(f"挂载失败: {result.stderr.strip()}")
        return True

    @classmethod
    def unmount(cls, mount_point: str, force: bool = False) -> bool:
        """卸载设备"""
        target = cls.MOUNT_BASE / mount_point.lstrip("/")
        cmd = ["umount"]
        if force:
            cmd.append("-f")
        cmd.append(str(target))

        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if result.returncode != 0:
            raise OSError(f"卸载失败: {result.stderr.strip()}")
        return True

    @classmethod
    def get_mounts(cls) -> list[dict]:
        """获取当前所有挂载点"""
        with open("/proc/mounts") as f:
            mounts = []
            for line in f:
                parts = line.strip().split()
                if len(parts) >= 4:
                    mounts.append({
                        "device": parts[0],
                        "mount_point": parts[1],
                        "fs_type": parts[2],
                        "options": parts[3],
                    })
            return mounts

11.2 模块 2：网络服务管理

Samba 共享管理器

"""nas/services/samba_manager.py — SMB 共享管理"""

import subprocess
import tempfile
import shutil
from pathlib import Path
from dataclasses import dataclass, field


@dataclass
class SMBShareConfig:
    """SMB 共享配置"""
    name: str
    path: str
    comment: str = ""
    browseable: bool = True
    read_only: bool = False
    guest_ok: bool = False
    valid_users: list[str] = field(default_factory=list)
    write_list: list[str] = field(default_factory=list)
    create_mask: str = "0664"
    directory_mask: str = "0775"


class SambaManager:
    """Samba 服务管理器"""

    SMB_CONF = Path("/etc/samba/smb.conf")
    BACKUP_DIR = Path("/etc/samba/backups")

    @classmethod
    def add_share(cls, config: SMBShareConfig) -> bool:
        """添加 SMB 共享"""
        # 验证共享名
        if not config.name or len(config.name) > 15:
            raise ValueError("共享名必须是 1-15 个字符")
        if not Path(config.path).exists():
            Path(config.path).mkdir(parents=True, exist_ok=True)

        # 生成配置段
        section = cls._generate_config_section(config)

        # 备份原配置
        cls._backup_config()

        # 追加配置
        with open(cls.SMB_CONF, 'a') as f:
            f.write(f"\n{section}\n")

        # 重载服务
        return cls.reload()

    @classmethod
    def remove_share(cls, name: str) -> bool:
        """删除 SMB 共享"""
        cls._backup_config()
        content = cls.SMB_CONF.read_text()
        # 移除 [share_name] 段（包括其所有配置行）
        pattern = rf'\n?\s*\[{re.escape(name)}\].*?(?=\n\s*\[|\Z)'
        content = re.sub(pattern, '', content, flags=re.DOTALL)
        cls.SMB_CONF.write_text(content)
        return cls.reload()

    @classmethod
    def list_shares(cls) -> list[dict]:
        """列出所有自定义共享"""
        content = cls.SMB_CONF.read_text()
        shares = []
        in_share = False
        current = {}

        for line in content.split("\n"):
            line = line.strip()
            if line.startswith("[") and line.endswith("]"):
                if in_share and current:
                    shares.append(current)
                name = line[1:-1]
                # 跳过全局段和默认共享
                if name not in ("global", "homes", "printers", "print$"):
                    in_share = True
                    current = {"name": name}
                else:
                    in_share = False
                    current = {}
            elif in_share and "=" in line:
                key, val = line.split("=", 1)
                current[key.strip()] = val.strip()

        if in_share and current:
            shares.append(current)
        return shares

    @classmethod
    def reload(cls) -> bool:
        """重载 Samba 配置"""
        result = subprocess.run(
            ["systemctl", "reload", "smbd"],
            capture_output=True, text=True, timeout=10
        )
        return result.returncode == 0

    @staticmethod
    def _generate_config_section(config: SMBShareConfig) -> str:
        lines = [
            f"[{config.name}]",
            f"  path = {config.path}",
        ]
        if config.comment:
            lines.append(f"  comment = {config.comment}")
        lines.append(f"  browseable = {'yes' if config.browseable else 'no'}")
        lines.append(f"  read only = {'yes' if config.read_only else 'no'}")
        lines.append(f"  guest ok = {'yes' if config.guest_ok else 'no'}")
        if config.valid_users:
            lines.append(f"  valid users = {' '.join(config.valid_users)}")
        if config.write_list:
            lines.append(f"  write list = {' '.join(config.write_list)}")
        lines.append(f"  create mask = {config.create_mask}")
        lines.append(f"  directory mask = {config.directory_mask}")
        return "\n".join(lines)

    @classmethod
    def _backup_config(cls):
        cls.BACKUP_DIR.mkdir(parents=True, exist_ok=True)
        import time
        stamp = time.strftime("%Y%m%d_%H%M%S")
        shutil.copy2(cls.SMB_CONF, cls.BACKUP_DIR / f"smb.conf.{stamp}")

11.3 模块 3：用户与权限系统

"""nas/auth/user_manager.py — 用户认证与权限管理"""

import subprocess
import pwd
import grp
import os
from dataclasses import dataclass
from typing import Optional


@dataclass
class UserInfo:
    username: str
    uid: int
    gid: int
    groups: list[str]
    home: str
    shell: str
    quota_gb: float = 0


class UserManager:
    """NAS 用户管理器 — PAM 认证 + 本地用户管理"""

    @staticmethod
    def list_users(min_uid: int = 1000) -> list[UserInfo]:
        """列出所有普通用户"""
        users = []
        for entry in pwd.getpwall():
            if entry.pw_uid >= min_uid and entry.pw_uid < 65534:
                groups = [
                    g.gr_name for g in grp.getgrall()
                    if entry.pw_name in g.gr_mem
                ]
                # 获取用户主组
                main_group = grp.getgrgid(entry.pw_gid)
                if main_group.gr_name not in groups:
                    groups.insert(0, main_group.gr_name)

                users.append(UserInfo(
                    username=entry.pw_name,
                    uid=entry.pw_uid,
                    gid=entry.pw_gid,
                    groups=groups,
                    home=entry.pw_dir,
                    shell=entry.pw_shell,
                ))
        return users

    @classmethod
    def create_user(cls, username: str, password: str,
                    groups: list[str] = None, shell: str = "/bin/bash") -> bool:
        """创建 NAS 用户"""
        # 验证用户名
        if not username or not username.isalnum():
            raise ValueError("用户名只能包含字母和数字")

        # 检查是否已存在
        try:
            pwd.getpwnam(username)
            raise ValueError(f"用户 {username} 已存在")
        except KeyError:
            pass

        # 创建用户
        home = f"/home/{username}"
        cmd = ["useradd", "-m", "-d", home, "-s", shell, username]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            raise RuntimeError(f"用户创建失败: {result.stderr}")

        # 设置密码
        set_pw = subprocess.run(
            ["chpasswd"],
            input=f"{username}:{password}",
            capture_output=True, text=True
        )
        if set_pw.returncode != 0:
            # 回滚
            subprocess.run(["userdel", "-r", username])
            raise RuntimeError(f"密码设置失败: {set_pw.stderr}")

        # 添加到附加组
        if groups:
            subprocess.run(["usermod", "-aG", ",".join(groups), username])

        return True

    @classmethod
    def delete_user(cls, username: str, remove_home: bool = True) -> bool:
        """删除用户"""
        # 防止删除系统用户
        entry = pwd.getpwnam(username)
        if entry.pw_uid < 1000:
            raise ValueError(f"不能删除系统用户: {username}")

        cmd = ["userdel"]
        if remove_home:
            cmd.append("-r")
        cmd.append(username)

        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.returncode == 0

    @classmethod
    def set_quota(cls, username: str, soft_gb: float, hard_gb: float,
                  filesystem: str = "/") -> bool:
        """设置用户磁盘配额"""
        soft_kb = int(soft_gb * 1024 * 1024)
        hard_kb = int(hard_gb * 1024 * 1024)

        result = subprocess.run(
            ["setquota", "-u", username,
             str(soft_kb), str(hard_kb), "0", "0", filesystem],
            capture_output=True, text=True
        )
        return result.returncode == 0

    @staticmethod
    def authenticate(username: str, password: str) -> bool:
        """通过 PAM 验证用户密码"""
        import pwd
        import crypt
        import spwd

        try:
            shadow = spwd.getspnam(username)
            hashed = shadow.sp_pwd if shadow.sp_pwd != "!" else None
            if not hashed:
                return False
            return crypt.crypt(password, hashed) == hashed
        except (KeyError, PermissionError):
            return False