marcandClaude Sonnet 4.6
0d09882268
SMB: force user, force group, guest ok fields + dropdowns for owner/group NFS: no_root_squash and all_squash toggles exposed in edit modal Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
754 lines
24 KiB
Python
754 lines
24 KiB
Python
import os
|
|
import re
|
|
import pwd
|
|
import grp
|
|
import shutil
|
|
import hashlib
|
|
import subprocess
|
|
import functools
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, request, jsonify, render_template, Response, session, redirect, url_for
|
|
|
|
app = Flask(__name__)
|
|
|
|
VALID_NAME = re.compile(r'^[A-Za-z0-9_-]+$')
|
|
VALID_NFS_CLIENT = re.compile(r'^[A-Za-z0-9.*:/\-]+$')
|
|
ALLOWED_PATH_BASE = Path('/srv').resolve()
|
|
|
|
_AUTH_USER = os.environ.get('MICRONAS_USER', 'admin')
|
|
_AUTH_PASS = os.environ.get('MICRONAS_PASS', '')
|
|
|
|
# Derive secret key from the password so sessions survive restarts
|
|
# but are automatically invalidated if the password changes.
|
|
app.secret_key = os.environ.get('MICRONAS_SECRET') or \
|
|
hashlib.sha256((_AUTH_PASS or 'insecure-dev-key').encode()).hexdigest()
|
|
|
|
if not _AUTH_PASS:
|
|
import warnings
|
|
warnings.warn(
|
|
'MICRONAS_PASS is not set — all routes are unprotected! '
|
|
'Set it via the environment or /etc/micronas.env.',
|
|
stacklevel=1,
|
|
)
|
|
|
|
|
|
def require_auth(f):
|
|
@functools.wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not _AUTH_PASS:
|
|
return f(*args, **kwargs)
|
|
if not session.get('logged_in'):
|
|
if request.path.startswith('/api/'):
|
|
return Response('Unauthorized', 401)
|
|
return redirect(url_for('login'))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
def validate_path(path_str):
|
|
try:
|
|
p = Path(path_str).resolve()
|
|
except Exception:
|
|
raise ValueError('invalid path')
|
|
allowed = str(ALLOWED_PATH_BASE)
|
|
if not (str(p) == allowed or str(p).startswith(allowed + '/')):
|
|
raise ValueError('path must be under /srv')
|
|
return str(p)
|
|
|
|
|
|
def _path_stat(path):
|
|
"""Return owner, group, and octal mode for a path. Empty strings on error."""
|
|
try:
|
|
s = os.stat(path)
|
|
return {
|
|
'owner': pwd.getpwuid(s.st_uid).pw_name,
|
|
'group': grp.getgrgid(s.st_gid).gr_name,
|
|
'mode': oct(s.st_mode)[-3:],
|
|
}
|
|
except Exception:
|
|
return {'owner': '', 'group': '', 'mode': ''}
|
|
|
|
|
|
def _validate_users(valid_users):
|
|
"""Validate and return a cleaned comma-separated user list, or raise ValueError."""
|
|
users = []
|
|
for u in re.split(r'[\s,]+', valid_users):
|
|
if not u:
|
|
continue
|
|
if not VALID_NAME.match(u):
|
|
raise ValueError(f'invalid username: {u}')
|
|
users.append(u)
|
|
return ', '.join(users)
|
|
|
|
|
|
# ===== SMB =====
|
|
|
|
_SMB_SKIP = {'global', 'homes', 'printers', 'print$', 'ipc$'}
|
|
|
|
|
|
def _parse_smb_conf():
|
|
"""Parse /etc/samba/smb.conf into sections. First entry (name=None) is the preamble."""
|
|
sections = []
|
|
current = {'name': None, 'lines': []}
|
|
try:
|
|
with open('/etc/samba/smb.conf', 'r') as f:
|
|
for line in f:
|
|
m = re.match(r'^\s*\[(.+)\]\s*$', line)
|
|
if m:
|
|
sections.append(current)
|
|
current = {'name': m.group(1).strip(), 'lines': [line]}
|
|
else:
|
|
current['lines'].append(line)
|
|
except FileNotFoundError:
|
|
return []
|
|
sections.append(current)
|
|
return sections
|
|
|
|
|
|
def _write_smb_conf(sections):
|
|
with open('/etc/samba/smb.conf', 'w') as f:
|
|
for s in sections:
|
|
f.writelines(s['lines'])
|
|
subprocess.run(['smbcontrol', 'all', 'reload-config'], capture_output=True)
|
|
|
|
|
|
def parse_smb():
|
|
result = subprocess.run(['testparm', '-s'], capture_output=True, text=True)
|
|
shares = []
|
|
current = None
|
|
|
|
for line in result.stdout.splitlines():
|
|
line = line.rstrip()
|
|
m = re.match(r'^\[(.+)\]$', line)
|
|
if m:
|
|
if current is not None and current['name'].lower() not in _SMB_SKIP:
|
|
shares.append(current)
|
|
current = {'name': m.group(1), 'path': '', 'valid_users': '',
|
|
'enabled': True, 'guest_ok': False,
|
|
'force_user': '', 'force_group': ''}
|
|
elif current is not None and '=' in line:
|
|
key, _, val = line.partition('=')
|
|
key = key.strip().lower().replace(' ', '_')
|
|
val = val.strip()
|
|
if key == 'path':
|
|
current['path'] = val
|
|
elif key == 'valid_users':
|
|
current['valid_users'] = val
|
|
elif key == 'available':
|
|
current['enabled'] = val.lower() not in ('no', 'false', '0')
|
|
elif key == 'guest_ok':
|
|
current['guest_ok'] = val.lower() in ('yes', 'true', '1')
|
|
elif key == 'force_user':
|
|
current['force_user'] = val
|
|
elif key == 'force_group':
|
|
current['force_group'] = val
|
|
|
|
if current is not None and current['name'].lower() not in _SMB_SKIP:
|
|
shares.append(current)
|
|
|
|
return shares
|
|
|
|
|
|
def add_smb_share(name, path, valid_users):
|
|
config = (
|
|
f'\n[{name}]\n'
|
|
f' path = {path}\n'
|
|
f' valid users = {valid_users}\n'
|
|
' writable = yes\n'
|
|
' browseable = yes\n'
|
|
)
|
|
with open('/etc/samba/smb.conf', 'a') as f:
|
|
f.write(config)
|
|
subprocess.run(['smbcontrol', 'all', 'reload-config'], check=True)
|
|
|
|
|
|
# ===== NFS =====
|
|
|
|
_DISABLED_PREFIX = '## ' # marker MicroNAS uses for disabled exports
|
|
|
|
|
|
def _parse_exports():
|
|
"""Parse /etc/exports. Only lines starting with '## /' are treated as
|
|
disabled exports — everything else that starts with '#' is a plain comment
|
|
and is left untouched."""
|
|
entries = []
|
|
try:
|
|
with open('/etc/exports', 'r') as f:
|
|
for raw in f:
|
|
stripped = raw.strip()
|
|
if not stripped:
|
|
entries.append({'kind': 'blank', 'raw': raw})
|
|
continue
|
|
if stripped.startswith(_DISABLED_PREFIX + '/'):
|
|
inner = stripped[len(_DISABLED_PREFIX):]
|
|
parts = inner.split(None, 1)
|
|
entries.append({'kind': 'export', 'enabled': False,
|
|
'path': parts[0], 'rest': parts[1] if len(parts) > 1 else ''})
|
|
continue
|
|
if stripped.startswith('#'):
|
|
entries.append({'kind': 'comment', 'raw': raw})
|
|
continue
|
|
parts = stripped.split(None, 1)
|
|
entries.append({'kind': 'export', 'enabled': True,
|
|
'path': parts[0], 'rest': parts[1] if len(parts) > 1 else ''})
|
|
except FileNotFoundError:
|
|
pass
|
|
return entries
|
|
|
|
|
|
def _write_exports(entries):
|
|
with open('/etc/exports', 'w') as f:
|
|
for e in entries:
|
|
if e['kind'] == 'export':
|
|
prefix = '' if e['enabled'] else _DISABLED_PREFIX
|
|
rest = f'\t{e["rest"]}' if e['rest'] else ''
|
|
f.write(f'{prefix}{e["path"]}{rest}\n')
|
|
else:
|
|
f.write(e['raw'])
|
|
subprocess.run(['exportfs', '-ra'], check=True)
|
|
|
|
|
|
def get_nfs_shares():
|
|
shares = []
|
|
for e in _parse_exports():
|
|
if e['kind'] != 'export':
|
|
continue
|
|
rest = e['rest']
|
|
shares.append({
|
|
'path': e['path'],
|
|
'clients': rest,
|
|
'enabled': e['enabled'],
|
|
'no_root_squash': 'no_root_squash' in rest,
|
|
'all_squash': 'all_squash' in rest,
|
|
})
|
|
return shares
|
|
|
|
|
|
def add_nfs_share(path, clients, ro=False):
|
|
options = 'ro' if ro else 'rw'
|
|
export_line = f'{path}\t{clients}({options},sync,no_subtree_check)\n'
|
|
with open('/etc/exports', 'a') as f:
|
|
f.write(export_line)
|
|
subprocess.run(['exportfs', '-ra'], check=True)
|
|
|
|
|
|
# ===== AUTH ROUTES =====
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if session.get('logged_in'):
|
|
return redirect(url_for('index'))
|
|
error = None
|
|
if request.method == 'POST':
|
|
username = request.form.get('username', '')
|
|
password = request.form.get('password', '')
|
|
if not _AUTH_PASS or (username == _AUTH_USER and password == _AUTH_PASS):
|
|
session['logged_in'] = True
|
|
return redirect(url_for('index'))
|
|
error = 'Invalid username or password.'
|
|
return render_template('login.html', error=error)
|
|
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for('login'))
|
|
|
|
|
|
# ===== MAIN ROUTES =====
|
|
|
|
@app.route('/')
|
|
@require_auth
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
|
|
@app.route('/api/shares', methods=['GET'])
|
|
@require_auth
|
|
def get_all_shares():
|
|
smb = parse_smb()
|
|
for s in smb:
|
|
s.update(_path_stat(s['path']))
|
|
nfs = get_nfs_shares()
|
|
for s in nfs:
|
|
s.update(_path_stat(s['path']))
|
|
return jsonify({'smb': smb, 'nfs': nfs})
|
|
|
|
|
|
# ===== SMB CRUD =====
|
|
|
|
@app.route('/api/share/smb', methods=['POST'])
|
|
@require_auth
|
|
def create_smb():
|
|
data = request.json or {}
|
|
name = data.get('name', '')
|
|
path = data.get('path', '')
|
|
valid_users = data.get('validUsers', '')
|
|
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid share name (letters, digits, _ -)'}), 400
|
|
try:
|
|
clean_users = _validate_users(valid_users)
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
add_smb_share(name, safe_path, clean_users)
|
|
return jsonify({'type': 'smb', 'status': 'created'})
|
|
|
|
|
|
@app.route('/api/share/smb/<name>', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_smb(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid share name'}), 400
|
|
sections = _parse_smb_conf()
|
|
filtered = [s for s in sections if s['name'] != name]
|
|
if len(filtered) == len(sections):
|
|
return jsonify({'error': f'share {name!r} not found'}), 404
|
|
_write_smb_conf(filtered)
|
|
return jsonify({'status': 'deleted'})
|
|
|
|
|
|
@app.route('/api/share/smb/<name>', methods=['PATCH'])
|
|
@require_auth
|
|
def edit_smb(name):
|
|
data = request.json or {}
|
|
new_name = data.get('name', name).strip()
|
|
path = data.get('path', '')
|
|
valid_users = data.get('validUsers', '')
|
|
force_user = data.get('forceUser', '').strip()
|
|
force_group = data.get('forceGroup', '').strip()
|
|
guest_ok = data.get('guestOk', False)
|
|
|
|
if not VALID_NAME.match(new_name):
|
|
return jsonify({'error': 'invalid share name'}), 400
|
|
if force_user and not VALID_NAME.match(force_user):
|
|
return jsonify({'error': 'invalid force user'}), 400
|
|
if force_group and not VALID_NAME.match(force_group):
|
|
return jsonify({'error': 'invalid force group'}), 400
|
|
try:
|
|
clean_users = _validate_users(valid_users)
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
lines = [
|
|
f'[{new_name}]\n',
|
|
f' path = {safe_path}\n',
|
|
f' valid users = {clean_users}\n',
|
|
' writable = yes\n',
|
|
' browseable = yes\n',
|
|
]
|
|
if force_user:
|
|
lines.append(f' force user = {force_user}\n')
|
|
if force_group:
|
|
lines.append(f' force group = {force_group}\n')
|
|
if guest_ok:
|
|
lines.append(' guest ok = yes\n')
|
|
|
|
sections = _parse_smb_conf()
|
|
for s in sections:
|
|
if s['name'] == name:
|
|
s['name'] = new_name
|
|
s['lines'] = lines
|
|
break
|
|
else:
|
|
return jsonify({'error': f'share {name!r} not found'}), 404
|
|
_write_smb_conf(sections)
|
|
return jsonify({'status': 'updated'})
|
|
|
|
|
|
@app.route('/api/share/smb/<name>/toggle', methods=['POST'])
|
|
@require_auth
|
|
def toggle_smb(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid share name'}), 400
|
|
enabled = (request.json or {}).get('enabled', True)
|
|
|
|
sections = _parse_smb_conf()
|
|
for s in sections:
|
|
if s['name'] == name:
|
|
s['lines'] = [l for l in s['lines'] if not re.match(r'\s*available\s*=', l, re.I)]
|
|
if not enabled:
|
|
s['lines'].append(' available = no\n')
|
|
break
|
|
else:
|
|
return jsonify({'error': f'share {name!r} not found'}), 404
|
|
_write_smb_conf(sections)
|
|
return jsonify({'status': 'enabled' if enabled else 'disabled'})
|
|
|
|
|
|
# ===== NFS CRUD =====
|
|
|
|
@app.route('/api/share/nfs', methods=['POST'])
|
|
@require_auth
|
|
def create_nfs():
|
|
data = request.json or {}
|
|
path = data.get('path', '')
|
|
clients = data.get('clients', '*')
|
|
ro = data.get('readOnly', False)
|
|
|
|
try:
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
if not VALID_NFS_CLIENT.match(clients):
|
|
return jsonify({'error': 'invalid client spec'}), 400
|
|
|
|
add_nfs_share(safe_path, clients, ro)
|
|
return jsonify({'type': 'nfs', 'status': 'created'})
|
|
|
|
|
|
@app.route('/api/share/nfs', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_nfs():
|
|
path = (request.json or {}).get('path', '')
|
|
try:
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
entries = _parse_exports()
|
|
original = sum(1 for e in entries if e['kind'] == 'export')
|
|
entries = [e for e in entries if not (e['kind'] == 'export' and e['path'] == safe_path)]
|
|
if sum(1 for e in entries if e['kind'] == 'export') == original:
|
|
return jsonify({'error': f'export {safe_path!r} not found'}), 404
|
|
_write_exports(entries)
|
|
return jsonify({'status': 'deleted'})
|
|
|
|
|
|
@app.route('/api/share/nfs', methods=['PATCH'])
|
|
@require_auth
|
|
def edit_nfs():
|
|
data = request.json or {}
|
|
path = data.get('path', '')
|
|
clients = data.get('clients', '*')
|
|
ro = data.get('readOnly', False)
|
|
|
|
try:
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
if not VALID_NFS_CLIENT.match(clients):
|
|
return jsonify({'error': 'invalid client spec'}), 400
|
|
|
|
no_root_squash = data.get('noRootSquash', False)
|
|
all_squash = data.get('allSquash', False)
|
|
options = ','.join(filter(None, [
|
|
'ro' if ro else 'rw',
|
|
'sync',
|
|
'no_subtree_check',
|
|
'no_root_squash' if no_root_squash else None,
|
|
'all_squash' if all_squash else None,
|
|
]))
|
|
new_rest = f'{clients}({options})'
|
|
|
|
entries = _parse_exports()
|
|
for e in entries:
|
|
if e['kind'] == 'export' and e['path'] == safe_path:
|
|
e['rest'] = new_rest
|
|
break
|
|
else:
|
|
return jsonify({'error': f'export {safe_path!r} not found'}), 404
|
|
_write_exports(entries)
|
|
return jsonify({'status': 'updated'})
|
|
|
|
|
|
@app.route('/api/share/nfs/toggle', methods=['POST'])
|
|
@require_auth
|
|
def toggle_nfs():
|
|
data = request.json or {}
|
|
path = data.get('path', '')
|
|
enabled = data.get('enabled', True)
|
|
|
|
try:
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
entries = _parse_exports()
|
|
for e in entries:
|
|
if e['kind'] == 'export' and e['path'] == safe_path:
|
|
e['enabled'] = enabled
|
|
break
|
|
else:
|
|
return jsonify({'error': f'export {safe_path!r} not found'}), 404
|
|
_write_exports(entries)
|
|
return jsonify({'status': 'enabled' if enabled else 'disabled'})
|
|
|
|
|
|
# ===== PERMISSIONS =====
|
|
|
|
@app.route('/api/permissions', methods=['POST'])
|
|
@require_auth
|
|
def set_permissions():
|
|
data = request.json or {}
|
|
path = data.get('path', '')
|
|
owner = data.get('owner', '').strip()
|
|
group = data.get('group', '').strip()
|
|
mode = data.get('mode', '').strip()
|
|
|
|
try:
|
|
safe_path = validate_path(path)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
if owner and not VALID_NAME.match(owner):
|
|
return jsonify({'error': 'invalid owner name'}), 400
|
|
if group and not VALID_NAME.match(group):
|
|
return jsonify({'error': 'invalid group name'}), 400
|
|
if mode and not re.match(r'^[0-7]{3,4}$', mode):
|
|
return jsonify({'error': 'mode must be 3-4 octal digits (e.g. 770)'}), 400
|
|
|
|
if owner or group:
|
|
chown_arg = f'{owner}:{group}' if (owner and group) else (owner or f':{group}')
|
|
r = subprocess.run(['chown', chown_arg, safe_path], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'chown failed'}), 500
|
|
|
|
if mode:
|
|
r = subprocess.run(['chmod', mode, safe_path], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'chmod failed'}), 500
|
|
|
|
return jsonify({'status': 'ok', 'stat': _path_stat(safe_path)})
|
|
|
|
|
|
# ===== GROUPS =====
|
|
|
|
_GROUP_WHITELIST = {'root', 'staff', 'sambashare', 'www-data'}
|
|
|
|
|
|
def _list_groups():
|
|
groups = []
|
|
with open('/etc/group', 'r') as f:
|
|
for line in f:
|
|
parts = line.strip().split(':')
|
|
if len(parts) < 4:
|
|
continue
|
|
name, gid_str, members_str = parts[0], parts[2], parts[3]
|
|
try:
|
|
gid = int(gid_str)
|
|
except ValueError:
|
|
continue
|
|
members = [m for m in members_str.split(',') if m]
|
|
if gid >= 1000 or name in _GROUP_WHITELIST:
|
|
groups.append({'name': name, 'gid': gid, 'members': members})
|
|
return sorted(groups, key=lambda g: g['name'])
|
|
|
|
|
|
@app.route('/api/groups', methods=['GET'])
|
|
@require_auth
|
|
def list_groups():
|
|
return jsonify(_list_groups())
|
|
|
|
|
|
@app.route('/api/groups', methods=['POST'])
|
|
@require_auth
|
|
def create_group():
|
|
name = (request.json or {}).get('name', '')
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid group name'}), 400
|
|
r = subprocess.run(['groupadd', name], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'groupadd failed'}), 400
|
|
return jsonify({'status': 'created'})
|
|
|
|
|
|
@app.route('/api/groups/<name>', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_group(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid group name'}), 400
|
|
r = subprocess.run(['groupdel', name], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'groupdel failed'}), 500
|
|
return jsonify({'status': 'deleted'})
|
|
|
|
|
|
@app.route('/api/groups/<name>/members', methods=['POST'])
|
|
@require_auth
|
|
def add_group_member(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid group name'}), 400
|
|
user = (request.json or {}).get('user', '')
|
|
if not VALID_NAME.match(user):
|
|
return jsonify({'error': 'invalid username'}), 400
|
|
r = subprocess.run(['gpasswd', '-a', user, name], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'gpasswd failed'}), 500
|
|
return jsonify({'status': 'added'})
|
|
|
|
|
|
@app.route('/api/groups/<name>/members/<user>', methods=['DELETE'])
|
|
@require_auth
|
|
def remove_group_member(name, user):
|
|
if not VALID_NAME.match(name) or not VALID_NAME.match(user):
|
|
return jsonify({'error': 'invalid name'}), 400
|
|
r = subprocess.run(['gpasswd', '-d', user, name], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'gpasswd failed'}), 500
|
|
return jsonify({'status': 'removed'})
|
|
|
|
|
|
# ===== USERS =====
|
|
|
|
def _smb_users():
|
|
try:
|
|
r = subprocess.run(['pdbedit', '-L'], capture_output=True, text=True)
|
|
return {line.split(':')[0].strip() for line in r.stdout.splitlines() if ':' in line}
|
|
except FileNotFoundError:
|
|
return set()
|
|
|
|
|
|
def _list_users():
|
|
smb = _smb_users()
|
|
users = []
|
|
with open('/etc/passwd', 'r') as f:
|
|
for line in f:
|
|
parts = line.strip().split(':')
|
|
if len(parts) < 4:
|
|
continue
|
|
name, uid = parts[0], parts[2]
|
|
uid = int(uid)
|
|
if uid < 1000 or uid == 65534:
|
|
continue
|
|
users.append({'name': name, 'uid': uid, 'smb': name in smb})
|
|
return users
|
|
|
|
|
|
def _set_smb_password(name, password):
|
|
r = subprocess.run(
|
|
['smbpasswd', '-a', '-s', name],
|
|
input=f'{password}\n{password}\n',
|
|
capture_output=True, text=True,
|
|
)
|
|
if r.returncode != 0:
|
|
raise RuntimeError(r.stderr.strip() or 'smbpasswd failed')
|
|
|
|
|
|
@app.route('/api/users', methods=['GET'])
|
|
@require_auth
|
|
def list_users():
|
|
return jsonify(_list_users())
|
|
|
|
|
|
@app.route('/api/users', methods=['POST'])
|
|
@require_auth
|
|
def create_user():
|
|
data = request.json or {}
|
|
name = data.get('name', '')
|
|
password = data.get('password', '')
|
|
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid username (letters, digits, _ -)'}), 400
|
|
if len(password) < 8:
|
|
return jsonify({'error': 'password must be at least 8 characters'}), 400
|
|
|
|
try:
|
|
subprocess.run(['useradd', '-M', '-s', '/usr/sbin/nologin', name], check=True)
|
|
except subprocess.CalledProcessError:
|
|
return jsonify({'error': f'user {name!r} already exists or useradd failed'}), 400
|
|
try:
|
|
_set_smb_password(name, password)
|
|
except RuntimeError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
return jsonify({'status': 'created'})
|
|
|
|
|
|
@app.route('/api/users/<name>', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_user(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid username'}), 400
|
|
subprocess.run(['smbpasswd', '-x', name], capture_output=True)
|
|
try:
|
|
subprocess.run(['userdel', name], check=True)
|
|
except subprocess.CalledProcessError as e:
|
|
return jsonify({'error': f'userdel failed: {e}'}), 500
|
|
return jsonify({'status': 'deleted'})
|
|
|
|
|
|
@app.route('/api/users/<name>/password', methods=['POST'])
|
|
@require_auth
|
|
def change_password(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid username'}), 400
|
|
data = request.json or {}
|
|
password = data.get('password', '')
|
|
if len(password) < 8:
|
|
return jsonify({'error': 'password must be at least 8 characters'}), 400
|
|
try:
|
|
_set_smb_password(name, password)
|
|
except RuntimeError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
return jsonify({'status': 'updated'})
|
|
|
|
|
|
# ===== DASHBOARD =====
|
|
|
|
def _daemon_status():
|
|
def is_active(service):
|
|
r = subprocess.run(['systemctl', 'is-active', service], capture_output=True, text=True)
|
|
return r.stdout.strip() == 'active'
|
|
return {'smbd': is_active('smbd'), 'nfsd': is_active('nfs-server')}
|
|
|
|
|
|
def _disk_usage():
|
|
try:
|
|
u = shutil.disk_usage('/srv')
|
|
return {
|
|
'path': '/srv',
|
|
'total_gb': round(u.total / 1024**3, 1),
|
|
'used_gb': round(u.used / 1024**3, 1),
|
|
'free_gb': round(u.free / 1024**3, 1),
|
|
'percent': round(u.used / u.total * 100, 1) if u.total else 0,
|
|
}
|
|
except Exception as e:
|
|
return {'error': str(e)}
|
|
|
|
|
|
def _du_bytes(path):
|
|
try:
|
|
r = subprocess.run(['du', '-sb', path], capture_output=True, text=True, timeout=15)
|
|
if r.returncode == 0:
|
|
parts = r.stdout.split()
|
|
return int(parts[0]) if parts else 0
|
|
except Exception:
|
|
pass
|
|
return 0
|
|
|
|
|
|
def _shares_with_usage():
|
|
result = []
|
|
for s in parse_smb():
|
|
result.append({
|
|
'type': 'smb', 'name': s['name'], 'path': s['path'],
|
|
'access': s['valid_users'], 'enabled': s['enabled'],
|
|
'size_gb': round(_du_bytes(s['path']) / 1024**3, 2) if s['path'] else 0,
|
|
})
|
|
for s in get_nfs_shares():
|
|
result.append({
|
|
'type': 'nfs', 'name': None, 'path': s['path'],
|
|
'access': s['clients'], 'enabled': s['enabled'],
|
|
'size_gb': round(_du_bytes(s['path']) / 1024**3, 2) if s['path'] else 0,
|
|
})
|
|
return result
|
|
|
|
|
|
|
|
@app.route('/api/dashboard', methods=['GET'])
|
|
@require_auth
|
|
def get_dashboard():
|
|
return jsonify({
|
|
'status': _daemon_status(),
|
|
'disk': _disk_usage(),
|
|
'shares': _shares_with_usage(),
|
|
})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000, debug=False)
|