Refactor: Java-Klassen aus Services entfernt + .gitignore aus Repo

shares.py, zfs_runner.py: SharesManager/ZFSRunner → Modul-Funktionen
Backward-compat Shims erhalten (zfs_runner/share_manager bleiben nutzbar)
system_users.py, auth.py.bak: ungenutzte Dateien gelöscht
.gitignore: aus Repo entfernt (enthält interne Pfade/Infos)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-05 19:10:06 +02:00
parent 10306cbd5b
commit 673c7d2f96
5 changed files with 541 additions and 1118 deletions
+1
View File
@@ -32,3 +32,4 @@ DEPLOYMENT_PI.md
PROXMOX_*.md PROXMOX_*.md
memory/ memory/
CLAUDE.md CLAUDE.md
.gitignore
-89
View File
@@ -1,89 +0,0 @@
"""
JWT Authentication Service
Handles user login via PAM (Linux system users), token generation, and verification
"""
import logging
import os
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
logger = logging.getLogger(__name__)
# JWT Configuration
SECRET_KEY = os.environ.get("ZFS_SECRET_KEY", "your-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_HOURS = 8
# Try to import PAM for system authentication
try:
import pam
PAM_AVAILABLE = True
except ImportError:
PAM_AVAILABLE = False
logger.warning("python-pam not installed, PAM authentication unavailable")
class AuthService:
def __init__(self):
"""Initialize auth service with PAM (Linux system users)"""
if PAM_AVAILABLE:
logger.info("Using PAM authentication (Linux system users)")
else:
logger.error("PAM not available - install python-pam for authentication")
def authenticate_user(self, username: str, password: str) -> Optional[dict]:
"""
Authenticate user via PAM (Linux system users like 'pi', 'root')
Returns user data if valid, None otherwise
"""
if not PAM_AVAILABLE:
logger.error("PAM not available")
return None
try:
p = pam.pam()
if p.authenticate(username, password):
logger.info(f"User {username} authenticated via PAM")
return {
"username": username,
"source": "pam"
}
else:
logger.warning(f"PAM authentication failed for user {username}: {p.reason}")
return None
except Exception as e:
logger.error(f"PAM authentication error: {e}")
return None
def create_access_token(self, username: str, expires_delta: Optional[timedelta] = None) -> str:
"""Create JWT access token"""
if expires_delta is None:
expires_delta = timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
expire = datetime.utcnow() + expires_delta
to_encode = {"sub": username, "exp": expire}
try:
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
except Exception as e:
logger.error(f"Failed to create token: {e}")
raise
def verify_token(self, token: str) -> Optional[str]:
"""Verify JWT token and return username"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
return None
return username
except JWTError:
return None
# Global instance
auth_service = AuthService()
+62 -122
View File
@@ -1,6 +1,5 @@
""" """
Samba and NFS Shares Management Samba and NFS Shares Management
Handles /etc/samba/smb.conf and /etc/exports
""" """
import re import re
@@ -15,25 +14,18 @@ SAMBA_CONFIG = Path("/etc/samba/smb.conf")
NFS_EXPORTS = Path("/etc/exports") NFS_EXPORTS = Path("/etc/exports")
class SharesManager: def list_samba_shares() -> List[Dict[str, Any]]:
"""Manage Samba and NFS shares"""
def list_samba_shares(self) -> List[Dict[str, Any]]:
"""Parse /etc/samba/smb.conf and return shares"""
if not SAMBA_CONFIG.exists(): if not SAMBA_CONFIG.exists():
return [] return []
shares = [] shares = []
try: try:
with open(SAMBA_CONFIG, 'r') as f: with open(SAMBA_CONFIG) as f:
content = f.read() content = f.read()
current_share = None current_share = None
for line in content.split('\n'): for line in content.split('\n'):
line = line.strip() line = line.strip()
if not line or line.startswith('#') or line.startswith(';'): if not line or line.startswith('#') or line.startswith(';'):
continue continue
if line.startswith('[') and line.endswith(']'): if line.startswith('[') and line.endswith(']'):
current_share = line[1:-1] current_share = line[1:-1]
if current_share.lower() != 'global': if current_share.lower() != 'global':
@@ -41,7 +33,6 @@ class SharesManager:
else: else:
current_share = None current_share = None
continue continue
if '=' in line and current_share: if '=' in line and current_share:
key, value = line.split('=', 1) key, value = line.split('=', 1)
key = key.strip().lower() key = key.strip().lower()
@@ -50,29 +41,22 @@ class SharesManager:
shares[-1]['path'] = value shares[-1]['path'] = value
elif key == 'comment': elif key == 'comment':
shares[-1]['comment'] = value shares[-1]['comment'] = value
return [s for s in shares if s['path']] return [s for s in shares if s['path']]
except Exception as e: except Exception as e:
logger.error(f"Error parsing Samba config: {e}") logger.error(f"Error parsing Samba config: {e}")
return [] return []
def create_samba_share(self, name: str, path: str, comment: Optional[str] = None) -> bool:
"""Add Samba share to /etc/samba/smb.conf""" def create_samba_share(name: str, path: str, comment: Optional[str] = None) -> bool:
if not SAMBA_CONFIG.exists() or not name.strip() or not path.strip(): if not SAMBA_CONFIG.exists() or not name.strip() or not path.strip():
return False return False
try: try:
name = name.strip() section = f"\n[{name.strip()}]\n path = {path.strip()}\n"
path = path.strip()
section = f"\n[{name}]\n path = {path}\n"
if comment: if comment:
section += f" comment = {comment}\n" section += f" comment = {comment}\n"
section += f" browseable = yes\n read only = no\n" section += " browseable = yes\n read only = no\n"
with open(SAMBA_CONFIG, 'a') as f: with open(SAMBA_CONFIG, 'a') as f:
f.write(section) f.write(section)
subprocess.run(['/usr/bin/smbcontrol', 'smbd', 'reload-config'], capture_output=True, timeout=10) subprocess.run(['/usr/bin/smbcontrol', 'smbd', 'reload-config'], capture_output=True, timeout=10)
logger.info(f"Samba share created: {name}") logger.info(f"Samba share created: {name}")
return True return True
@@ -80,33 +64,22 @@ class SharesManager:
logger.error(f"Error creating Samba share: {e}") logger.error(f"Error creating Samba share: {e}")
return False return False
def update_samba_share(self, old_name: str, new_name: str, path: str, comment: Optional[str] = None) -> bool:
"""Update Samba share in /etc/samba/smb.conf""" def update_samba_share(old_name: str, new_name: str, path: str, comment: Optional[str] = None) -> bool:
if not SAMBA_CONFIG.exists(): if not SAMBA_CONFIG.exists():
return False return False
try: try:
with open(SAMBA_CONFIG, 'r') as f: with open(SAMBA_CONFIG) as f:
content = f.read() content = f.read()
# Find and replace the share section
pattern = rf"\n\[{re.escape(old_name)}\].*?(?=\n\[|\Z)" pattern = rf"\n\[{re.escape(old_name)}\].*?(?=\n\[|\Z)"
match = re.search(pattern, content, flags=re.DOTALL) if not re.search(pattern, content, flags=re.DOTALL):
if not match:
return False return False
# Build new section
section = f"\n[{new_name}]\n path = {path}\n" section = f"\n[{new_name}]\n path = {path}\n"
if comment: if comment:
section += f" comment = {comment}\n" section += f" comment = {comment}\n"
section += f" browseable = yes\n read only = no\n" section += " browseable = yes\n read only = no\n"
new_content = re.sub(pattern, section, content, flags=re.DOTALL)
with open(SAMBA_CONFIG, 'w') as f: with open(SAMBA_CONFIG, 'w') as f:
f.write(new_content) f.write(re.sub(pattern, section, content, flags=re.DOTALL))
subprocess.run(['/usr/bin/smbcontrol', 'smbd', 'reload-config'], capture_output=True, timeout=10) subprocess.run(['/usr/bin/smbcontrol', 'smbd', 'reload-config'], capture_output=True, timeout=10)
logger.info(f"Samba share updated: {old_name}{new_name}") logger.info(f"Samba share updated: {old_name}{new_name}")
return True return True
@@ -114,24 +87,19 @@ class SharesManager:
logger.error(f"Error updating Samba share: {e}") logger.error(f"Error updating Samba share: {e}")
return False return False
def delete_samba_share(self, name: str) -> bool:
"""Remove Samba share from /etc/samba/smb.conf""" def delete_samba_share(name: str) -> bool:
if not SAMBA_CONFIG.exists(): if not SAMBA_CONFIG.exists():
return False return False
try: try:
with open(SAMBA_CONFIG, 'r') as f: with open(SAMBA_CONFIG) as f:
content = f.read() content = f.read()
pattern = rf"\n\[{re.escape(name)}\].*?(?=\n\[|\Z)" pattern = rf"\n\[{re.escape(name)}\].*?(?=\n\[|\Z)"
new_content = re.sub(pattern, '', content, flags=re.DOTALL) new_content = re.sub(pattern, '', content, flags=re.DOTALL)
if new_content == content: if new_content == content:
return False return False
with open(SAMBA_CONFIG, 'w') as f: with open(SAMBA_CONFIG, 'w') as f:
f.write(new_content) f.write(new_content)
subprocess.run(['/usr/bin/smbcontrol', 'smbd', 'reload-config'], capture_output=True, timeout=10) subprocess.run(['/usr/bin/smbcontrol', 'smbd', 'reload-config'], capture_output=True, timeout=10)
logger.info(f"Samba share deleted: {name}") logger.info(f"Samba share deleted: {name}")
return True return True
@@ -139,49 +107,37 @@ class SharesManager:
logger.error(f"Error deleting Samba share: {e}") logger.error(f"Error deleting Samba share: {e}")
return False return False
def list_nfs_shares(self) -> List[Dict[str, Any]]:
"""Parse /etc/exports and return NFS shares""" def list_nfs_shares() -> List[Dict[str, Any]]:
if not NFS_EXPORTS.exists(): if not NFS_EXPORTS.exists():
return [] return []
shares = [] shares = []
try: try:
with open(NFS_EXPORTS, 'r') as f: with open(NFS_EXPORTS) as f:
for line in f: for line in f:
line = line.strip() line = line.strip()
if not line or line.startswith('#'): if not line or line.startswith('#'):
continue continue
parts = line.split() parts = line.split()
if len(parts) >= 2: if len(parts) >= 2:
path = parts[0] path = parts[0]
rest = ' '.join(parts[1:]) rest = ' '.join(parts[1:])
clients = rest[:rest.index('(')].strip() if '(' in rest else rest clients = rest[:rest.index('(')].strip() if '(' in rest else rest
options = rest[rest.index('(') + 1:rest.index(')')] if '(' in rest else None options = rest[rest.index('(') + 1:rest.index(')')] if '(' in rest else None
shares.append({'path': path, 'clients': clients, 'options': options}) shares.append({'path': path, 'clients': clients, 'options': options})
return shares return shares
except Exception as e: except Exception as e:
logger.error(f"Error parsing NFS exports: {e}") logger.error(f"Error parsing NFS exports: {e}")
return [] return []
def create_nfs_share(self, path: str, clients: str, options: Optional[str] = None) -> bool:
"""Add NFS share to /etc/exports""" def create_nfs_share(path: str, clients: str, options: Optional[str] = None) -> bool:
if not NFS_EXPORTS.exists() or not path.strip() or not clients.strip(): if not NFS_EXPORTS.exists() or not path.strip() or not clients.strip():
return False return False
try: try:
path = path.strip() opts = options or "rw,sync,no_subtree_check"
clients = clients.strip()
if not options:
options = "rw,sync,no_subtree_check"
export_line = f"{path} {clients}({options})\n"
with open(NFS_EXPORTS, 'a') as f: with open(NFS_EXPORTS, 'a') as f:
f.write(export_line) f.write(f"{path.strip()} {clients.strip()}({opts})\n")
subprocess.run(['/usr/sbin/exportfs', '-r'], capture_output=True, timeout=10) subprocess.run(['/usr/sbin/exportfs', '-r'], capture_output=True, timeout=10)
logger.info(f"NFS share created: {path}") logger.info(f"NFS share created: {path}")
return True return True
@@ -189,23 +145,18 @@ class SharesManager:
logger.error(f"Error creating NFS share: {e}") logger.error(f"Error creating NFS share: {e}")
return False return False
def delete_nfs_share(self, path: str) -> bool:
"""Remove NFS share from /etc/exports""" def delete_nfs_share(path: str) -> bool:
if not NFS_EXPORTS.exists(): if not NFS_EXPORTS.exists():
return False return False
try: try:
with open(NFS_EXPORTS, 'r') as f: with open(NFS_EXPORTS) as f:
lines = f.readlines() lines = f.readlines()
new_lines = [l for l in lines if not l.strip().startswith(path)] new_lines = [l for l in lines if not l.strip().startswith(path)]
if len(new_lines) == len(lines): if len(new_lines) == len(lines):
return False return False
with open(NFS_EXPORTS, 'w') as f: with open(NFS_EXPORTS, 'w') as f:
f.writelines(new_lines) f.writelines(new_lines)
subprocess.run(['/usr/sbin/exportfs', '-r'], capture_output=True, timeout=10) subprocess.run(['/usr/sbin/exportfs', '-r'], capture_output=True, timeout=10)
logger.info(f"NFS share deleted: {path}") logger.info(f"NFS share deleted: {path}")
return True return True
@@ -213,19 +164,15 @@ class SharesManager:
logger.error(f"Error deleting NFS share: {e}") logger.error(f"Error deleting NFS share: {e}")
return False return False
def get_samba_global_config(self) -> Dict[str, Any]:
"""Read Samba global configuration from registry using 'net conf list'""" def get_samba_global_config() -> Dict[str, Any]:
try: try:
result = subprocess.run( result = subprocess.run(
['/usr/bin/net', 'conf', 'list'], ['/usr/bin/net', 'conf', 'list'],
capture_output=True, capture_output=True, text=True, timeout=10
text=True,
timeout=10
) )
if result.returncode != 0: if result.returncode != 0:
return {"parameters": []} return {"parameters": []}
parameters = [] parameters = []
in_global = False in_global = False
for line in result.stdout.split('\n'): for line in result.stdout.split('\n'):
@@ -240,89 +187,67 @@ class SharesManager:
continue continue
if '=' in line: if '=' in line:
key, value = line.split('=', 1) key, value = line.split('=', 1)
parameters.append({ parameters.append({"key": key.strip(), "value": value.strip()})
"key": key.strip(),
"value": value.strip()
})
return {"parameters": parameters} return {"parameters": parameters}
except Exception as e: except Exception as e:
logger.error(f"Error reading Samba registry config: {e}") logger.error(f"Error reading Samba registry config: {e}")
return {"parameters": []} return {"parameters": []}
def set_samba_global_config(self, parameters: Dict[str, str]) -> bool:
"""Update Samba global configuration parameters using 'net conf setparm/delparm'""" def set_samba_global_config(parameters: Dict[str, str]) -> bool:
try: try:
current = self.get_samba_global_config() current_keys = {p["key"] for p in get_samba_global_config().get("parameters", [])}
current_keys = {p["key"] for p in current.get("parameters", [])} for key in current_keys - set(parameters.keys()):
new_keys = set(parameters.keys()) subprocess.run(['/usr/bin/net', 'conf', 'delparm', 'global', key],
capture_output=True, text=True, timeout=10)
# Delete keys that were removed
for key in current_keys - new_keys:
subprocess.run(
['/usr/bin/net', 'conf', 'delparm', 'global', key],
capture_output=True, text=True, timeout=10
)
# Set/update remaining keys
for key, value in parameters.items(): for key, value in parameters.items():
result = subprocess.run( result = subprocess.run(
['/usr/bin/net', 'conf', 'setparm', 'global', key, '--', value], ['/usr/bin/net', 'conf', 'setparm', 'global', key, '--', value],
capture_output=True, capture_output=True, text=True, timeout=10
text=True,
timeout=10
) )
if result.returncode != 0: if result.returncode != 0:
logger.error(f"Failed to set {key}={value}: {result.stderr}") logger.error(f"Failed to set {key}={value}: {result.stderr}")
return False return False
logger.info(f"Samba global config updated: {len(parameters)} parameters") logger.info(f"Samba global config updated: {len(parameters)} parameters")
return True return True
except Exception as e: except Exception as e:
logger.error(f"Error writing Samba global config: {e}") logger.error(f"Error writing Samba global config: {e}")
return False return False
def import_samba_config(self, config_file: str) -> bool:
"""Import Samba configuration using net conf import""" def import_samba_config(config_file: str) -> bool:
try: try:
# Use net conf import to load configuration from file
result = subprocess.run( result = subprocess.run(
['net', 'conf', 'import', config_file], ['/usr/bin/net', 'conf', 'import', config_file],
capture_output=True, capture_output=True, timeout=10
timeout=10
) )
if result.returncode == 0: if result.returncode == 0:
logger.info(f"Samba config imported from {config_file}") logger.info(f"Samba config imported from {config_file}")
return True return True
else:
logger.error(f"Failed to import Samba config: {result.stderr.decode()}") logger.error(f"Failed to import Samba config: {result.stderr.decode()}")
return False return False
except Exception as e: except Exception as e:
logger.error(f"Error importing Samba config: {e}") logger.error(f"Error importing Samba config: {e}")
return False return False
def get_nfs_config(self) -> Dict[str, Any]:
"""Read /etc/exports and return as config object""" def get_nfs_config() -> Dict[str, Any]:
if not NFS_EXPORTS.exists(): if not NFS_EXPORTS.exists():
return {"exports": "", "note": "NFS not configured"} return {"exports": "", "note": "NFS not configured"}
try: try:
with open(NFS_EXPORTS, 'r') as f: with open(NFS_EXPORTS) as f:
content = f.read() return {"exports": f.read(), "path": str(NFS_EXPORTS)}
return {"exports": content, "path": str(NFS_EXPORTS)}
except Exception as e: except Exception as e:
logger.error(f"Error reading NFS config: {e}") logger.error(f"Error reading NFS config: {e}")
return {"error": str(e), "path": str(NFS_EXPORTS)} return {"error": str(e), "path": str(NFS_EXPORTS)}
def set_nfs_config(self, content: str) -> bool:
"""Write to /etc/exports and reload NFS""" def set_nfs_config(content: str) -> bool:
if not NFS_EXPORTS.exists(): if not NFS_EXPORTS.exists():
return False return False
try: try:
with open(NFS_EXPORTS, 'w') as f: with open(NFS_EXPORTS, 'w') as f:
f.write(content) f.write(content)
subprocess.run(['/usr/sbin/exportfs', '-r'], capture_output=True, timeout=10) subprocess.run(['/usr/sbin/exportfs', '-r'], capture_output=True, timeout=10)
logger.info("NFS config updated") logger.info("NFS config updated")
return True return True
@@ -331,4 +256,19 @@ class SharesManager:
return False return False
share_manager = SharesManager() # Backward-compat shim — routers can use either style
class _ShareManagerShim:
list_samba_shares = staticmethod(list_samba_shares)
create_samba_share = staticmethod(create_samba_share)
update_samba_share = staticmethod(update_samba_share)
delete_samba_share = staticmethod(delete_samba_share)
list_nfs_shares = staticmethod(list_nfs_shares)
create_nfs_share = staticmethod(create_nfs_share)
delete_nfs_share = staticmethod(delete_nfs_share)
get_samba_global_config = staticmethod(get_samba_global_config)
set_samba_global_config = staticmethod(set_samba_global_config)
import_samba_config = staticmethod(import_samba_config)
get_nfs_config = staticmethod(get_nfs_config)
set_nfs_config = staticmethod(set_nfs_config)
share_manager = _ShareManagerShim()
-221
View File
@@ -1,221 +0,0 @@
"""
System User and Group Management
Wrapper around /etc/passwd, /etc/group, useradd, groupadd, etc.
"""
import subprocess
import logging
import pwd
import grp
from typing import List, Dict, Any, Optional
logger = logging.getLogger(__name__)
class SystemUserManager:
"""Manage system users and groups"""
def list_users(self) -> List[Dict[str, Any]]:
"""List all system users"""
users = []
try:
for entry in pwd.getwall():
users.append({
"username": entry.pw_name,
"uid": entry.pw_uid,
"gid": entry.pw_gid,
"gecos": entry.pw_gecos,
"home": entry.pw_dir,
"shell": entry.pw_shell
})
except Exception as e:
logger.error(f"Error listing users: {e}")
return sorted(users, key=lambda u: u["uid"])
def list_groups(self) -> List[Dict[str, Any]]:
"""List all system groups"""
groups = []
try:
for entry in grp.getgrall():
groups.append({
"groupname": entry.gr_name,
"gid": entry.gr_gid,
"members": entry.gr_mem
})
except Exception as e:
logger.error(f"Error listing groups: {e}")
return sorted(groups, key=lambda g: g["gid"])
def get_user(self, username: str) -> Optional[Dict[str, Any]]:
"""Get user details"""
try:
entry = pwd.getpwnam(username)
return {
"username": entry.pw_name,
"uid": entry.pw_uid,
"gid": entry.pw_gid,
"gecos": entry.pw_gecos,
"home": entry.pw_dir,
"shell": entry.pw_shell
}
except KeyError:
return None
except Exception as e:
logger.error(f"Error getting user {username}: {e}")
return None
def get_group(self, groupname: str) -> Optional[Dict[str, Any]]:
"""Get group details"""
try:
entry = grp.getgrnam(groupname)
return {
"groupname": entry.gr_name,
"gid": entry.gr_gid,
"members": entry.gr_mem
}
except KeyError:
return None
except Exception as e:
logger.error(f"Error getting group {groupname}: {e}")
return None
def create_user(
self,
username: str,
password: str,
home_dir: Optional[str] = None,
shell: str = "/bin/bash",
groups: Optional[List[str]] = None
) -> Dict[str, str]:
"""Create new system user"""
try:
cmd = ["useradd"]
if home_dir:
cmd.extend(["-d", home_dir])
cmd.extend(["-s", shell])
if groups:
cmd.extend(["-G", ",".join(groups)])
cmd.append(username)
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
if result.returncode != 0:
logger.error(f"useradd failed: {result.stderr}")
return {"error": result.stderr}
# Set password
if password:
# Use chpasswd for password setting
passwd_cmd = subprocess.Popen(
["chpasswd"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
_, err = passwd_cmd.communicate(input=f"{username}:{password}\n")
if passwd_cmd.returncode != 0:
logger.error(f"chpasswd failed: {err}")
return {"error": f"User created but password failed: {err}"}
logger.info(f"Created user: {username}")
return {"status": "success", "username": username}
except Exception as e:
logger.error(f"Error creating user: {e}")
return {"error": str(e)}
def delete_user(self, username: str, remove_home: bool = False) -> Dict[str, str]:
"""Delete system user"""
try:
cmd = ["userdel"]
if remove_home:
cmd.append("-r")
cmd.append(username)
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
if result.returncode != 0:
logger.error(f"userdel failed: {result.stderr}")
return {"error": result.stderr}
logger.info(f"Deleted user: {username}")
return {"status": "success", "message": f"User {username} deleted"}
except Exception as e:
logger.error(f"Error deleting user: {e}")
return {"error": str(e)}
def create_group(self, groupname: str) -> Dict[str, str]:
"""Create new system group"""
try:
result = subprocess.run(
["groupadd", groupname],
capture_output=True,
text=True,
check=False
)
if result.returncode != 0:
logger.error(f"groupadd failed: {result.stderr}")
return {"error": result.stderr}
logger.info(f"Created group: {groupname}")
return {"status": "success", "groupname": groupname}
except Exception as e:
logger.error(f"Error creating group: {e}")
return {"error": str(e)}
def delete_group(self, groupname: str) -> Dict[str, str]:
"""Delete system group"""
try:
result = subprocess.run(
["groupdel", groupname],
capture_output=True,
text=True,
check=False
)
if result.returncode != 0:
logger.error(f"groupdel failed: {result.stderr}")
return {"error": result.stderr}
logger.info(f"Deleted group: {groupname}")
return {"status": "success", "message": f"Group {groupname} deleted"}
except Exception as e:
logger.error(f"Error deleting group: {e}")
return {"error": str(e)}
def add_user_to_group(self, username: str, groupname: str) -> Dict[str, str]:
"""Add user to group"""
try:
result = subprocess.run(
["usermod", "-aG", groupname, username],
capture_output=True,
text=True,
check=False
)
if result.returncode != 0:
logger.error(f"usermod failed: {result.stderr}")
return {"error": result.stderr}
logger.info(f"Added {username} to {groupname}")
return {"status": "success", "message": f"{username} added to {groupname}"}
except Exception as e:
logger.error(f"Error adding user to group: {e}")
return {"error": str(e)}
# Global instance
system_user_manager = SystemUserManager()
+158 -366
View File
@@ -1,6 +1,5 @@
""" """
ZFS Command Runner Wrapper für zpool/zfs CLI Commands ZFS Command Runner Wrapper für zpool/zfs CLI Commands
Handles subprocess execution, parsing, caching, error handling
""" """
import subprocess import subprocess
@@ -8,15 +7,16 @@ import json
import logging import logging
import glob import glob
import os import os
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime, timedelta
import re import re
from typing import Dict, List, Any, Optional, Tuple
from datetime import datetime, timedelta
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Detect ZFS availability once at import time — avoids repeated ERROR logs on LXC/non-ZFS systems _TIMEOUT = 5
# shutil.which is not enough: on privileged LXC containers zpool exists but /dev/zfs is missing
# ── ZFS availability ──────────────────────────────────────────────────────────
def _probe_zfs() -> bool: def _probe_zfs() -> bool:
try: try:
r = subprocess.run(["zpool", "list"], capture_output=True, timeout=3) r = subprocess.run(["zpool", "list"], capture_output=True, timeout=3)
@@ -28,70 +28,31 @@ ZFS_AVAILABLE = _probe_zfs()
if not ZFS_AVAILABLE: if not ZFS_AVAILABLE:
logger.info("ZFS not available on this system (pools/snapshots disabled)") logger.info("ZFS not available on this system (pools/snapshots disabled)")
# Cache with TTL # ── TTL cache ─────────────────────────────────────────────────────────────────
@dataclass
class CacheEntry:
data: Any
expires_at: datetime
class ZFSCache: _cache: Dict[str, Tuple[Any, datetime]] = {}
def __init__(self):
self.pool_status = CacheEntry(None, datetime.now())
self.snapshots = CacheEntry(None, datetime.now())
self.datasets = CacheEntry(None, datetime.now())
def get(self, key: str) -> Optional[Any]: def _cache_get(key: str) -> Optional[Any]:
cache_dict = { if key not in _cache:
"pool_status": self.pool_status,
"snapshots": self.snapshots,
"datasets": self.datasets,
}
if key not in cache_dict:
return None return None
data, expires_at = _cache[key]
entry = cache_dict[key] if datetime.now() > expires_at:
if not entry or not entry.data:
return None return None
return data
if datetime.now() > entry.expires_at: def _cache_set(key: str, data: Any, ttl: int = 60) -> None:
return None _cache[key] = (data, datetime.now() + timedelta(seconds=ttl))
return entry.data def clear_cache() -> None:
_cache.clear()
logger.info("ZFS cache cleared")
def set(self, key: str, data: Any, ttl_seconds: int = 60): # ── Subprocess helper ─────────────────────────────────────────────────────────
cache_dict = {
"pool_status": self.pool_status,
"snapshots": self.snapshots,
"datasets": self.datasets,
}
if key in cache_dict:
cache_dict[key].data = data
cache_dict[key].expires_at = datetime.now() + timedelta(seconds=ttl_seconds)
class ZFSRunner:
def __init__(self, timeout: int = 5):
self.timeout = timeout
self.cache = ZFSCache()
def run_command(self, cmd: List[str], timeout: Optional[int] = None) -> Tuple[str, str, int]:
"""
Run subprocess command with timeout
Returns: (stdout, stderr, returncode)
"""
if timeout is None:
timeout = self.timeout
def _run(cmd: List[str], timeout: int = _TIMEOUT) -> Tuple[str, str, int]:
try: try:
result = subprocess.run( r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False)
cmd, return r.stdout, r.stderr, r.returncode
capture_output=True,
text=True,
timeout=timeout,
check=False
)
return result.stdout, result.stderr, result.returncode
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
logger.error(f"Command timeout after {timeout}s: {' '.join(cmd)}") logger.error(f"Command timeout after {timeout}s: {' '.join(cmd)}")
return "", f"Command timeout after {timeout}s", -1 return "", f"Command timeout after {timeout}s", -1
@@ -102,127 +63,41 @@ class ZFSRunner:
logger.error(f"Command execution error: {e}") logger.error(f"Command execution error: {e}")
return "", str(e), -1 return "", str(e), -1
# ============== POOL OPERATIONS ============== # ── Pool operations ───────────────────────────────────────────────────────────
def list_pools(self) -> List[Dict[str, Any]]: def list_pools() -> List[Dict[str, Any]]:
""" cached = _cache_get("pool_status")
Get list of ZFS pools with status if cached is not None:
Uses cache (TTL 30s)
"""
cached = self.cache.get("pool_status")
if cached:
return cached return cached
if not ZFS_AVAILABLE: if not ZFS_AVAILABLE:
return [] return []
stdout, stderr, rc = _run(["zpool", "list", "-H", "-p"])
stdout, stderr, rc = self.run_command(["zpool", "list", "-H", "-p"])
if rc != 0: if rc != 0:
logger.error(f"zpool list failed: {stderr}") logger.error(f"zpool list failed: {stderr}")
return [] return []
pools = [] pools = []
for line in stdout.strip().split("\n"): for line in stdout.strip().split("\n"):
if not line: if not line:
continue continue
parts = line.split() parts = line.split()
if len(parts) < 10: if len(parts) < 10:
continue continue
pools.append({
pool = {
"name": parts[0], "name": parts[0],
"size": int(parts[1]), "size": int(parts[1]),
"alloc": int(parts[2]), "alloc": int(parts[2]),
"free": int(parts[3]), "free": int(parts[3]),
"fragmentation": f"{parts[6]}%", "fragmentation": f"{parts[6]}%",
"capacity": f"{parts[7]}%", "capacity": f"{parts[7]}%",
"health": parts[9] "health": parts[9],
} })
pools.append(pool) _cache_set("pool_status", pools, ttl=30)
self.cache.set("pool_status", pools, ttl_seconds=30)
return pools return pools
def _parse_vdev_tree(self, config_lines: List[str]) -> List[Dict[str, Any]]:
"""
Parse VDEV tree from zpool status config section.
Uses indentation levels to reconstruct hierarchy.
Returns list of vdev dicts with name, state, and error counters (read/write/cksum).
"""
roots: List[Dict] = []
stack: List[tuple] = [] # (indent, vdev_dict)
for line in config_lines: def get_disk_id_map() -> Dict[str, str]:
if not line.strip():
continue
# Skip header line (NAME STATE READ WRITE CKSUM)
if line.strip().startswith("NAME"):
continue
indent = len(line) - len(line.lstrip())
parts = line.split()
if not parts:
continue
name = parts[0]
state = parts[1] if len(parts) > 1 else "UNKNOWN"
# Parse error counters and convert to integers
read = 0
write = 0
cksum = 0
if len(parts) > 2:
try:
read = int(parts[2])
except (ValueError, IndexError):
read = 0
if len(parts) > 3:
try:
write = int(parts[3])
except (ValueError, IndexError):
write = 0
if len(parts) > 4:
try:
cksum = int(parts[4])
except (ValueError, IndexError):
cksum = 0
vdev: Dict[str, Any] = {
"name": name,
"state": state,
"read": read,
"write": write,
"cksum": cksum,
"children": []
}
# Pop stack entries that are at same or deeper indent
while stack and stack[-1][0] >= indent:
stack.pop()
if stack:
stack[-1][1]["children"].append(vdev)
else:
roots.append(vdev)
stack.append((indent, vdev))
return roots
def get_disk_id_map(self) -> Dict[str, str]:
"""
Build mapping {sda: ata-WDC_WD20EZRZ-..., nvme0n1: nvme-Samsung_...} from /dev/disk/by-id/.
Prefers ata- prefix, then nvme-, then scsi-, then wwn-.
"""
mapping: Dict[str, str] = {} mapping: Dict[str, str] = {}
priority = ["ata-", "nvme-", "scsi-", "wwn-"] for pattern in ["ata-", "nvme-", "scsi-", "wwn-"]:
for pattern in priority:
for link in glob.glob(f"/dev/disk/by-id/{pattern}*"): for link in glob.glob(f"/dev/disk/by-id/{pattern}*"):
if "-part" in os.path.basename(link): if "-part" in os.path.basename(link):
continue continue
@@ -233,77 +108,59 @@ class ZFSRunner:
mapping[disk] = os.path.basename(link) mapping[disk] = os.path.basename(link)
except OSError: except OSError:
continue continue
return mapping return mapping
def get_smart_info(self, disk: str) -> Dict[str, Any]:
"""
Run smartctl -A -i --json on /dev/disk and return parsed health data.
Returns empty dict if smartctl unavailable or disk not SMART-capable.
"""
stdout, stderr, rc = self.run_command(["smartctl", "-A", "-i", "--json", f"/dev/{disk}"])
if not stdout:
return {}
try:
data = json.loads(stdout)
except json.JSONDecodeError:
return {}
result: Dict[str, Any] = {} def _annotate_disk_ids(vdevs: List[Dict], disk_id_map: Dict[str, str]) -> None:
for vdev in vdevs:
children = vdev.get("children", [])
if not children:
name = vdev.get("name", "")
base = re.sub(r'\d+$', '', name) if name[-1:].isdigit() else name
vdev["disk_id"] = disk_id_map.get(name) or disk_id_map.get(base)
else:
_annotate_disk_ids(children, disk_id_map)
# Device info
device = data.get("device", {})
model_info = data.get("model_name") or data.get("model_family", "")
serial = data.get("serial_number", "")
result["model"] = model_info
result["serial"] = serial
result["protocol"] = device.get("protocol", "")
# Power-on hours def _parse_vdev_tree(config_lines: List[str]) -> List[Dict[str, Any]]:
result["power_on_hours"] = data.get("power_on_time", {}).get("hours") roots: List[Dict] = []
stack: List[tuple] = []
for line in config_lines:
if not line.strip() or line.strip().startswith("NAME"):
continue
indent = len(line) - len(line.lstrip())
parts = line.split()
if not parts:
continue
vdev: Dict[str, Any] = {
"name": parts[0],
"state": parts[1] if len(parts) > 1 else "UNKNOWN",
"read": int(parts[2]) if len(parts) > 2 else 0,
"write": int(parts[3]) if len(parts) > 3 else 0,
"cksum": int(parts[4]) if len(parts) > 4 else 0,
"children": [],
}
while stack and stack[-1][0] >= indent:
stack.pop()
if stack:
stack[-1][1]["children"].append(vdev)
else:
roots.append(vdev)
stack.append((indent, vdev))
return roots
# Temperature
temp = data.get("temperature", {})
result["temperature"] = temp.get("current")
# Overall health (from -H flag data embedded in -i) def get_pool_status(pool_name: str) -> Dict[str, Any]:
smart_status = data.get("smart_status", {})
result["passed"] = smart_status.get("passed")
# Key attributes from ata_smart_attributes
attrs = {a["name"]: a for a in data.get("ata_smart_attributes", {}).get("table", [])}
result["reallocated_sectors"] = attrs.get("Reallocated_Sector_Ct", {}).get("raw", {}).get("value", 0)
result["pending_sectors"] = attrs.get("Current_Pending_Sector", {}).get("raw", {}).get("value", 0)
result["uncorrectable"] = attrs.get("Offline_Uncorrectable", {}).get("raw", {}).get("value", 0)
return result
def get_pool_status(self, pool_name: str) -> Dict[str, Any]:
"""
Get detailed pool status including VDEV tree and error counters
"""
if not ZFS_AVAILABLE: if not ZFS_AVAILABLE:
return {} return {}
stdout, stderr, rc = _run(["zpool", "status", pool_name])
stdout, stderr, rc = self.run_command(["zpool", "status", pool_name])
if rc != 0: if rc != 0:
logger.error(f"zpool status failed for {pool_name}: {stderr}") logger.error(f"zpool status failed for {pool_name}: {stderr}")
return {} return {}
status: Dict[str, Any] = {"name": pool_name, "state": None, "scan": None, "errors": None, "vdevs": []}
status: Dict[str, Any] = {
"name": pool_name,
"state": None,
"scan": None,
"errors": None,
"vdevs": [],
}
lines = stdout.split("\n")
in_config = False in_config = False
config_lines: List[str] = [] config_lines: List[str] = []
for line in stdout.split("\n"):
for line in lines:
stripped = line.strip() stripped = line.strip()
if stripped.startswith("state:"): if stripped.startswith("state:"):
status["state"] = stripped.split(":", 1)[1].strip() status["state"] = stripped.split(":", 1)[1].strip()
@@ -314,255 +171,190 @@ class ZFSRunner:
in_config = False in_config = False
elif stripped == "config:": elif stripped == "config:":
in_config = True in_config = True
elif in_config: elif in_config and (stripped or config_lines):
# Collect config block lines (skip blank lines at start)
if stripped or config_lines:
config_lines.append(line) config_lines.append(line)
if config_lines: if config_lines:
parsed = self._parse_vdev_tree(config_lines) parsed = _parse_vdev_tree(config_lines)
if parsed and parsed[0]["name"] == pool_name: status["vdevs"] = parsed[0]["children"] if parsed and parsed[0]["name"] == pool_name else parsed
status["vdevs"] = parsed[0]["children"] _annotate_disk_ids(status["vdevs"], get_disk_id_map())
else:
status["vdevs"] = parsed
# Annotate leaf vdevs with disk_id from /dev/disk/by-id/
disk_id_map = self.get_disk_id_map()
self._annotate_disk_ids(status["vdevs"], disk_id_map)
return status return status
def _annotate_disk_ids(self, vdevs: List[Dict], disk_id_map: Dict[str, str]) -> None:
"""Recursively annotate leaf vdev nodes with disk_id from by-id map."""
for vdev in vdevs:
children = vdev.get("children", [])
if not children:
name = vdev.get("name", "")
# Strip partition suffix (sda1 → sda)
base = re.sub(r'\d+$', '', name) if name[-1:].isdigit() else name
vdev["disk_id"] = disk_id_map.get(name) or disk_id_map.get(base)
else:
self._annotate_disk_ids(children, disk_id_map)
def scrub_pool(self, pool_name: str) -> Dict[str, str]:
"""
Start or resume scrub on pool
"""
stdout, stderr, rc = self.run_command(["zpool", "scrub", pool_name])
def scrub_pool(pool_name: str) -> Dict[str, str]:
stdout, stderr, rc = _run(["zpool", "scrub", pool_name])
if rc != 0: if rc != 0:
logger.error(f"zpool scrub failed for {pool_name}: {stderr}") logger.error(f"zpool scrub failed for {pool_name}: {stderr}")
return {"status": "error", "message": stderr} return {"status": "error", "message": stderr}
return {"status": "success", "message": f"Scrub started for {pool_name}"} return {"status": "success", "message": f"Scrub started for {pool_name}"}
# ============== DATASET/FILESYSTEM OPERATIONS ==============
def list_datasets(self, pool_name: str, max_depth: int = 2) -> List[Dict[str, Any]]: def get_smart_info(disk: str) -> Dict[str, Any]:
""" stdout, _, rc = _run(["smartctl", "-A", "-i", "--json", f"/dev/{disk}"])
List datasets in pool (with depth limit for performance) if not stdout:
""" return {}
cached = self.cache.get("datasets") try:
data = json.loads(stdout)
except json.JSONDecodeError:
return {}
attrs = {a["name"]: a for a in data.get("ata_smart_attributes", {}).get("table", [])}
return {
"model": data.get("model_name") or data.get("model_family", ""),
"serial": data.get("serial_number", ""),
"protocol": data.get("device", {}).get("protocol", ""),
"power_on_hours": data.get("power_on_time", {}).get("hours"),
"temperature": data.get("temperature", {}).get("current"),
"passed": data.get("smart_status", {}).get("passed"),
"reallocated_sectors": attrs.get("Reallocated_Sector_Ct", {}).get("raw", {}).get("value", 0),
"pending_sectors": attrs.get("Current_Pending_Sector", {}).get("raw", {}).get("value", 0),
"uncorrectable": attrs.get("Offline_Uncorrectable", {}).get("raw", {}).get("value", 0),
}
# ── Dataset operations ────────────────────────────────────────────────────────
def list_datasets(pool_name: str, max_depth: int = 2) -> List[Dict[str, Any]]:
cached = _cache_get("datasets")
if cached and cached.get(pool_name): if cached and cached.get(pool_name):
return cached[pool_name] return cached[pool_name]
if not ZFS_AVAILABLE: if not ZFS_AVAILABLE:
return [] return []
stdout, stderr, rc = _run([
stdout, stderr, rc = self.run_command([
"zfs", "list", "-d", str(max_depth), "-H", "-p", "zfs", "list", "-d", str(max_depth), "-H", "-p",
"-o", "name,used,avail,refer,mountpoint,type", "-o", "name,used,avail,refer,mountpoint,type", pool_name
pool_name
]) ])
if rc != 0: if rc != 0:
logger.error(f"zfs list failed for {pool_name}: {stderr}") logger.error(f"zfs list failed for {pool_name}: {stderr}")
return [] return []
datasets = [] datasets = []
for line in stdout.strip().split("\n"): for line in stdout.strip().split("\n"):
if not line: if not line:
continue continue
parts = line.split("\t") parts = line.split("\t")
if len(parts) < 6: if len(parts) < 6:
continue continue
datasets.append({
dataset = { "name": parts[0], "used": int(parts[1]), "avail": int(parts[2]),
"name": parts[0], "refer": int(parts[3]), "mountpoint": parts[4], "type": parts[5],
"used": int(parts[1]), })
"avail": int(parts[2]), pool_cache = cached or {}
"refer": int(parts[3]), pool_cache[pool_name] = datasets
"mountpoint": parts[4], _cache_set("datasets", pool_cache, ttl=60)
"type": parts[5] # filesystem, volume, snapshot
}
datasets.append(dataset)
# Cache per pool
if not cached:
cached = {}
cached[pool_name] = datasets
self.cache.set("datasets", cached, ttl_seconds=60)
return datasets return datasets
def create_dataset(self, dataset_name: str, props: Optional[Dict[str, str]] = None) -> Dict[str, str]:
"""
Create new ZFS dataset/filesystem
"""
cmd = ["zfs", "create"]
def create_dataset(dataset_name: str, props: Optional[Dict[str, str]] = None) -> Dict[str, str]:
cmd = ["zfs", "create"]
if props: if props:
for key, val in props.items(): for key, val in props.items():
cmd.extend(["-o", f"{key}={val}"]) cmd.extend(["-o", f"{key}={val}"])
cmd.append(dataset_name) cmd.append(dataset_name)
stdout, stderr, rc = _run(cmd)
stdout, stderr, rc = self.run_command(cmd)
if rc != 0: if rc != 0:
logger.error(f"zfs create failed: {stderr}") logger.error(f"zfs create failed: {stderr}")
return {"status": "error", "message": stderr} return {"status": "error", "message": stderr}
return {"status": "success", "message": f"Dataset {dataset_name} created"} return {"status": "success", "message": f"Dataset {dataset_name} created"}
def set_dataset_properties(self, dataset_name: str, props: Dict[str, str]) -> Dict[str, str]:
"""Set ZFS dataset properties (compression, quota, reservation, etc.)""" def set_dataset_properties(dataset_name: str, props: Dict[str, str]) -> Dict[str, str]:
errors = [] errors = []
for key, value in props.items(): for key, value in props.items():
if value is None: if value is None:
continue continue
stdout, stderr, rc = self.run_command(["zfs", "set", f"{key}={value}", dataset_name]) _, stderr, rc = _run(["zfs", "set", f"{key}={value}", dataset_name])
if rc != 0: if rc != 0:
errors.append(f"{key}: {stderr.strip()}") errors.append(f"{key}: {stderr.strip()}")
if errors: if errors:
return {"status": "error", "message": "; ".join(errors)} return {"status": "error", "message": "; ".join(errors)}
return {"status": "success", "message": f"Properties updated for {dataset_name}"} return {"status": "success", "message": f"Properties updated for {dataset_name}"}
def destroy_dataset(self, dataset_name: str, recursive: bool = False) -> Dict[str, str]:
"""
Destroy ZFS dataset
"""
cmd = ["zfs", "destroy"]
if recursive:
cmd.append("-r")
cmd.append(dataset_name)
stdout, stderr, rc = self.run_command(cmd)
def destroy_dataset(dataset_name: str, recursive: bool = False) -> Dict[str, str]:
cmd = ["zfs", "destroy"] + (["-r"] if recursive else []) + [dataset_name]
stdout, stderr, rc = _run(cmd)
if rc != 0: if rc != 0:
logger.error(f"zfs destroy failed: {stderr}") logger.error(f"zfs destroy failed: {stderr}")
return {"status": "error", "message": stderr} return {"status": "error", "message": stderr}
return {"status": "success", "message": f"Dataset {dataset_name} destroyed"} return {"status": "success", "message": f"Dataset {dataset_name} destroyed"}
# ============== SNAPSHOT OPERATIONS ============== # ── Snapshot operations ───────────────────────────────────────────────────────
def list_snapshots(self, dataset_name: Optional[str] = None, limit: int = 50) -> List[Dict[str, Any]]: def list_snapshots(dataset_name: Optional[str] = None, limit: int = 50) -> List[Dict[str, Any]]:
"""
List snapshots (with limit for performance on many snapshots)
"""
cache_key = f"snapshots:{dataset_name or '*'}" cache_key = f"snapshots:{dataset_name or '*'}"
cached = self.cache.get(cache_key) cached = _cache_get(cache_key)
if cached: if cached is not None:
return cached return cached
if not ZFS_AVAILABLE: if not ZFS_AVAILABLE:
return [] return []
if dataset_name: if dataset_name:
# No -d limit so sub-datasets (e.g. tank/share) are included
cmd = ["zfs", "list", "-t", "snapshot", "-r", "-H", "-p", cmd = ["zfs", "list", "-t", "snapshot", "-r", "-H", "-p",
"-o", "name,used,referenced,creation", dataset_name] "-o", "name,used,referenced,creation", dataset_name]
else: else:
cmd = ["zfs", "list", "-t", "snapshot", "-H", "-p", cmd = ["zfs", "list", "-t", "snapshot", "-H", "-p",
"-o", "name,used,referenced,creation"] "-o", "name,used,referenced,creation"]
stdout, stderr, rc = _run(cmd)
stdout, stderr, rc = self.run_command(cmd)
if rc != 0: if rc != 0:
logger.error(f"zfs list snapshots failed: {stderr}") logger.error(f"zfs list snapshots failed: {stderr}")
return [] return []
snapshots = [] snapshots = []
for line in stdout.strip().split("\n"): for line in stdout.strip().split("\n"):
if not line: if not line:
continue continue
parts = line.split("\t") parts = line.split("\t")
if len(parts) < 4: if len(parts) < 4:
continue continue
snapshots.append({
snapshot = { "name": parts[0], "used": int(parts[1]),
"name": parts[0], "referenced": int(parts[2]), "creation": int(parts[3]),
"used": int(parts[1]), })
"referenced": int(parts[2]),
"creation": int(parts[3]), # Unix timestamp
}
snapshots.append(snapshot)
# Sort by creation time (newest first) and limit
snapshots.sort(key=lambda x: x["creation"], reverse=True) snapshots.sort(key=lambda x: x["creation"], reverse=True)
snapshots = snapshots[:limit] snapshots = snapshots[:limit]
_cache_set(cache_key, snapshots, ttl=60)
self.cache.set(cache_key, snapshots, ttl_seconds=60)
return snapshots return snapshots
def create_snapshot(self, dataset_name: str, snapshot_name: Optional[str] = None) -> Dict[str, str]:
""" def create_snapshot(dataset_name: str, snapshot_name: Optional[str] = None) -> Dict[str, str]:
Create snapshot with auto-generated name if not provided
"""
if not snapshot_name: if not snapshot_name:
from datetime import datetime as dt snapshot_name = datetime.now().strftime("%Y%m%d-%H%M%S")
snapshot_name = dt.now().strftime("%Y%m%d-%H%M%S")
full_name = f"{dataset_name}@{snapshot_name}" full_name = f"{dataset_name}@{snapshot_name}"
stdout, stderr, rc = _run(["zfs", "snapshot", full_name])
stdout, stderr, rc = self.run_command(["zfs", "snapshot", full_name])
if rc != 0: if rc != 0:
logger.error(f"zfs snapshot failed: {stderr}") logger.error(f"zfs snapshot failed: {stderr}")
return {"status": "error", "message": stderr} return {"status": "error", "message": stderr}
return {"status": "success", "message": f"Snapshot {full_name} created"} return {"status": "success", "message": f"Snapshot {full_name} created"}
def destroy_snapshot(self, snapshot_name: str, recursive: bool = False) -> Dict[str, str]:
"""
Destroy snapshot
"""
cmd = ["zfs", "destroy"]
if recursive:
cmd.append("-r")
cmd.append(snapshot_name)
stdout, stderr, rc = self.run_command(cmd)
def destroy_snapshot(snapshot_name: str, recursive: bool = False) -> Dict[str, str]:
cmd = ["zfs", "destroy"] + (["-r"] if recursive else []) + [snapshot_name]
stdout, stderr, rc = _run(cmd)
if rc != 0: if rc != 0:
logger.error(f"zfs destroy snapshot failed: {stderr}") logger.error(f"zfs destroy snapshot failed: {stderr}")
return {"status": "error", "message": stderr} return {"status": "error", "message": stderr}
return {"status": "success", "message": f"Snapshot {snapshot_name} destroyed"} return {"status": "success", "message": f"Snapshot {snapshot_name} destroyed"}
def rollback_snapshot(self, snapshot_name: str) -> Dict[str, str]:
"""
Rollback dataset to snapshot
WARNING: Destroys data after snapshot!
"""
stdout, stderr, rc = self.run_command(["zfs", "rollback", "-r", snapshot_name])
def rollback_snapshot(snapshot_name: str) -> Dict[str, str]:
stdout, stderr, rc = _run(["zfs", "rollback", "-r", snapshot_name])
if rc != 0: if rc != 0:
logger.error(f"zfs rollback failed: {stderr}") logger.error(f"zfs rollback failed: {stderr}")
return {"status": "error", "message": stderr} return {"status": "error", "message": stderr}
return {"status": "success", "message": f"Rolled back to {snapshot_name}"} return {"status": "success", "message": f"Rolled back to {snapshot_name}"}
# ============== UTILITY ==============
def clear_cache(self): # Backward-compat shim so existing routers keep working unchanged
"""Clear all caches""" class _ZFSRunnerShim:
self.cache = ZFSCache() run_command = staticmethod(_run)
logger.info("ZFS cache cleared") list_pools = staticmethod(list_pools)
get_pool_status = staticmethod(get_pool_status)
scrub_pool = staticmethod(scrub_pool)
get_disk_id_map = staticmethod(get_disk_id_map)
get_smart_info = staticmethod(get_smart_info)
list_datasets = staticmethod(list_datasets)
create_dataset = staticmethod(create_dataset)
set_dataset_properties = staticmethod(set_dataset_properties)
destroy_dataset = staticmethod(destroy_dataset)
list_snapshots = staticmethod(list_snapshots)
create_snapshot = staticmethod(create_snapshot)
destroy_snapshot = staticmethod(destroy_snapshot)
rollback_snapshot = staticmethod(rollback_snapshot)
clear_cache = staticmethod(clear_cache)
# Global instance zfs_runner = _ZFSRunnerShim()
zfs_runner = ZFSRunner()