Template
- 用户管理:增删改、批量导入(Excel模板)、admin保护 - 部门管理:增删改(含回收站恢复机制)、删除保护(有用户不可删) - 权限管理:部门内角色(admin/readwrite/readonly/none) - 监控页面:CPU/内存/磁盘实时监控 - 日志页面:操作日志查询 - SMB权限:用户文件夹700(member)/750(admin)、ACL继承 - admin跨目录访问:通过setfacl ACL实现
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
"""Linux 系统用户和文件夹权限管理服务"""
|
||
import os
|
||
import subprocess
|
||
from app.utils.helpers import run_cmd
|
||
|
||
DATA_DIR = os.environ.get("DATA_DIR", "/data/smb")
|
||
|
||
|
||
def create_linux_user(username: str, password: str) -> dict:
|
||
"""创建 Linux 系统用户(无登录shell)"""
|
||
# 先检查是否已存在
|
||
rc, stdout, _ = run_cmd(["id", username])
|
||
if rc == 0:
|
||
return {"success": True, "exists": True}
|
||
|
||
# 创建用户(无home目录,无登录shell)
|
||
rc, _, stderr = run_cmd([
|
||
"useradd", "-M", "-s", "/usr/sbin/nologin", username
|
||
])
|
||
if rc != 0:
|
||
return {"success": False, "error": f"useradd failed: {stderr}"}
|
||
|
||
# 设置密码
|
||
proc = subprocess.run(
|
||
["chpasswd"],
|
||
input=f"{username}:{password}",
|
||
capture_output=True, text=True
|
||
)
|
||
if proc.returncode != 0:
|
||
# 回滚用户创建
|
||
run_cmd(["userdel", username])
|
||
return {"success": False, "error": "password set failed"}
|
||
|
||
return {"success": True, "exists": False}
|
||
|
||
|
||
def delete_linux_user(username: str) -> dict:
|
||
"""删除 Linux 系统用户"""
|
||
rc, _, _ = run_cmd(["id", username])
|
||
if rc != 0:
|
||
return {"success": True} # 用户不存在,视为成功
|
||
|
||
rc, _, stderr = run_cmd(["userdel", username])
|
||
if rc != 0:
|
||
return {"success": False, "error": f"userdel failed: {stderr}"}
|
||
return {"success": True}
|
||
|
||
|
||
def add_user_to_group(username: str, group_name: str) -> bool:
|
||
"""将用户加入 Linux 组"""
|
||
rc, _, _ = run_cmd(["usermod", "-aG", group_name, username])
|
||
return rc == 0
|
||
|
||
|
||
def remove_user_from_group(username: str, group_name: str) -> bool:
|
||
"""将用户从 Linux 组移除"""
|
||
rc, _, _ = run_cmd(["gpasswd", "-d", username, group_name])
|
||
return rc == 0
|
||
|
||
|
||
def create_department_group(group_name: str) -> dict:
|
||
"""创建部门 Linux 组"""
|
||
rc, _, _ = run_cmd(["groupadd", "-f", group_name])
|
||
return {"success": rc == 0 or rc == 4} # 4=组已存在
|
||
|
||
|
||
def create_user_folder(username: str, dept_name: str, role: str = "member") -> dict:
|
||
"""创建用户的个人文件夹"""
|
||
folder_path = os.path.join(DATA_DIR, f"dept-{dept_name}", username)
|
||
try:
|
||
os.makedirs(folder_path, exist_ok=True)
|
||
# 设置所有者
|
||
os.chown(folder_path, -1, -1) # 待Samba挂载后修正
|
||
os.chmod(folder_path, 0o700 if role == "member" else 0o750)
|
||
return {"success": True, "path": folder_path}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
def set_folder_owner(username: str, folder_path: str, group_name: str = None) -> bool:
|
||
"""设置文件夹所有权"""
|
||
import pwd
|
||
try:
|
||
uid = pwd.getpwnam(username).pw_uid
|
||
gid = pwd.getpwnam(username).pw_gid if not group_name else -1
|
||
if group_name:
|
||
import grp
|
||
gid = grp.getgrnam(group_name).gr_gid
|
||
os.chown(folder_path, uid, gid)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def create_department_folders(dept_name: str) -> dict:
|
||
"""创建部门根文件夹"""
|
||
dept_path = os.path.join(DATA_DIR, f"dept-{dept_name}")
|
||
admin_path = os.path.join(dept_path, "admin")
|
||
try:
|
||
os.makedirs(admin_path, exist_ok=True)
|
||
# admin 文件夹允许部门管理员访问
|
||
os.chmod(dept_path, 0o750)
|
||
os.chmod(admin_path, 0o700)
|
||
return {"success": True, "dept_path": dept_path, "admin_path": admin_path}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
def apply_posix_acl(folder_path: str, username: str, perms: str = "rwx") -> bool:
|
||
"""设置 POSIX ACL 权限"""
|
||
rc, _, _ = run_cmd(["setfacl", "-m", f"u:{username}:{perms}", folder_path])
|
||
return rc == 0
|