70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
"""YakPanel - FTP service (Pure-FTPd via pure-pw)"""
|
|
import os
|
|
import subprocess
|
|
from app.core.config import get_runtime_config
|
|
|
|
|
|
def _run_pure_pw(args: str, stdin: str | None = None) -> tuple[str, str]:
|
|
"""Run pure-pw command. Returns (stdout, stderr)."""
|
|
try:
|
|
r = subprocess.run(
|
|
f"pure-pw {args}",
|
|
shell=True,
|
|
input=stdin,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
return r.stdout or "", r.stderr or ""
|
|
except FileNotFoundError:
|
|
return "", "pure-pw not found"
|
|
except subprocess.TimeoutExpired:
|
|
return "", "Timed out"
|
|
|
|
|
|
def create_ftp_user(name: str, password: str, path: str) -> tuple[bool, str]:
|
|
"""Create Pure-FTPd virtual user via pure-pw."""
|
|
if not all(c.isalnum() or c in "._-" for c in name):
|
|
return False, "Invalid username"
|
|
if ".." in path:
|
|
return False, "Invalid path"
|
|
path_abs = os.path.abspath(path)
|
|
cfg = get_runtime_config()
|
|
www_root = os.path.abspath(cfg["www_root"])
|
|
if not (path_abs == www_root or path_abs.startswith(www_root + os.sep)):
|
|
return False, "Path must be under www_root"
|
|
os.makedirs(path_abs, exist_ok=True)
|
|
# pure-pw useradd prompts for password twice; pipe it
|
|
stdin = f"{password}\n{password}\n"
|
|
out, err = _run_pure_pw(
|
|
f'useradd {name} -u www-data -d "{path_abs}" -m',
|
|
stdin=stdin,
|
|
)
|
|
if err and "error" in err.lower() and "already exists" not in err.lower():
|
|
return False, err.strip() or out.strip()
|
|
out2, err2 = _run_pure_pw("mkdb")
|
|
if err2 and "error" in err2.lower():
|
|
return False, err2.strip() or out2.strip()
|
|
return True, "FTP user created"
|
|
|
|
|
|
def delete_ftp_user(name: str) -> tuple[bool, str]:
|
|
"""Delete Pure-FTPd virtual user."""
|
|
if not all(c.isalnum() or c in "._-" for c in name):
|
|
return False, "Invalid username"
|
|
out, err = _run_pure_pw(f'userdel {name} -m')
|
|
if err and "error" in err.lower():
|
|
return False, err.strip() or out.strip()
|
|
return True, "FTP user deleted"
|
|
|
|
|
|
def update_ftp_password(name: str, new_password: str) -> tuple[bool, str]:
|
|
"""Change Pure-FTPd user password."""
|
|
if not all(c.isalnum() or c in "._-" for c in name):
|
|
return False, "Invalid username"
|
|
stdin = f"{new_password}\n{new_password}\n"
|
|
out, err = _run_pure_pw(f'passwd {name} -m', stdin=stdin)
|
|
if err and "error" in err.lower():
|
|
return False, err.strip() or out.strip()
|
|
return True, "Password updated"
|