Files
MicroNAS/app.py
T
marcandClaude Sonnet 4.6 302a5cb023 Add folder permission editor to share edit modal
Backend:
- Import pwd/grp for uid/gid → name resolution
- _path_stat(path): returns current owner, group, octal mode via os.stat
- get_all_shares: attaches stat info to each share so the modal can
  pre-fill current values
- POST /api/permissions: validates and applies chown/chmod to a share
  path; returns updated stat on success

Frontend:
- Edit modal now has a Permissions section (Owner / Group / Mode)
  pre-filled with current values, with a hint line showing common modes
- Valid Users label notes @groupname syntax for Samba group access
- submitEdit applies permissions after saving the share config if any
  permission field is non-empty; shows specific error on failure

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-20 00:26:00 +02:00

637 lines
19 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}
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')
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():
return [
{'path': e['path'], 'clients': e['rest'], 'enabled': e['enabled']}
for e in _parse_exports() if e['kind'] == 'export'
]
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', '')
if not VALID_NAME.match(new_name):
return jsonify({'error': 'invalid share name'}), 400
try:
clean_users = _validate_users(valid_users)
safe_path = validate_path(path)
except ValueError as e:
return jsonify({'error': str(e)}), 400
sections = _parse_smb_conf()
for s in sections:
if s['name'] == name:
s['name'] = new_name
s['lines'] = [
f'[{new_name}]\n',
f' path = {safe_path}\n',
f' valid users = {clean_users}\n',
' writable = yes\n',
' browseable = yes\n',
]
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
options = 'ro' if ro else 'rw'
new_rest = f'{clients}({options},sync,no_subtree_check)'
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)})
# ===== 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)