marcandClaude Sonnet 4.6
9f060660a1
app.py: - Move warnings import to top-level imports - Remove MICRONAS_SECRET env var (undocumented, unnecessary) - Inline secret key derivation cleanly CLAUDE.md: - Remove stale quotas mention - Add WebDAV, setup.html, favicon to file layout - Document /etc/micronas/ permissions requirements - Document <Location> vs <Directory> for WebDAV DAV On PLANNING.md: - Mark Phase 5 complete with all shipped items - Add v1 complete marker and future ideas section README.md: - Simplify dev/test section (remove complex hash snippet) - Fix double blank line Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1089 lines
36 KiB
Python
1089 lines
36 KiB
Python
import os
|
|
import re
|
|
import pwd
|
|
import grp
|
|
import hmac
|
|
import shutil
|
|
import hashlib
|
|
import warnings
|
|
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()
|
|
|
|
_CONF_DIR = '/etc/micronas'
|
|
_ENV_FILE = '/etc/micronas/micronas.env'
|
|
|
|
# Mutable credential store — updated by setup/change without restart (single gunicorn worker).
|
|
_creds = {
|
|
'user': os.environ.get('MICRONAS_USER', ''),
|
|
'hash': os.environ.get('MICRONAS_PASS_HASH', ''), # PBKDF2: "salt$hex"
|
|
'plain': os.environ.get('MICRONAS_PASS', ''), # legacy plaintext
|
|
}
|
|
|
|
def _is_configured():
|
|
return bool(_creds['user'] and (_creds['hash'] or _creds['plain']))
|
|
|
|
def _hash_password(password):
|
|
salt = os.urandom(16).hex()
|
|
h = hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 260_000).hex()
|
|
return f'{salt}${h}'
|
|
|
|
def _verify_password(password):
|
|
if _creds['hash']:
|
|
try:
|
|
salt, h = _creds['hash'].split('$', 1)
|
|
computed = hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 260_000).hex()
|
|
return hmac.compare_digest(computed, h)
|
|
except Exception:
|
|
return False
|
|
if _creds['plain']:
|
|
return hmac.compare_digest(password, _creds['plain'])
|
|
return False
|
|
|
|
def _save_credentials(user, password):
|
|
h = _hash_password(password)
|
|
_ensure_conf_dir()
|
|
Path(_ENV_FILE).write_text(f'MICRONAS_USER={user}\nMICRONAS_PASS_HASH={h}\n')
|
|
os.chmod(_ENV_FILE, 0o600)
|
|
_creds['user'] = user
|
|
_creds['hash'] = h
|
|
_creds['plain'] = ''
|
|
app.secret_key = hashlib.sha256(h.encode()).hexdigest()
|
|
|
|
# Derive initial secret key from stored credentials
|
|
_secret_src = _creds['hash'] or _creds['plain'] or 'insecure-dev-key'
|
|
app.secret_key = hashlib.sha256(_secret_src.encode()).hexdigest()
|
|
|
|
if not _is_configured():
|
|
warnings.warn('No credentials configured — first-time setup required at /setup', stacklevel=1)
|
|
|
|
|
|
def require_auth(f):
|
|
@functools.wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not _is_configured():
|
|
if request.path.startswith('/api/'):
|
|
return Response('Setup required', 403)
|
|
return redirect(url_for('setup'))
|
|
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': ''}
|
|
|
|
|
|
_VALID_SMB_ENTRY = re.compile(r'^@?[A-Za-z0-9_-]+$') # allows @groupname for SMB
|
|
|
|
|
|
def _validate_users(valid_users):
|
|
"""Validate and return a cleaned comma-separated user list (SMB: allows @group)."""
|
|
users = []
|
|
for u in re.split(r'[\s,]+', valid_users):
|
|
if not u:
|
|
continue
|
|
if not _VALID_SMB_ENTRY.match(u):
|
|
raise ValueError(f'invalid username: {u!r}')
|
|
users.append(u)
|
|
return ', '.join(users)
|
|
|
|
|
|
def _validate_webdav_users(valid_users):
|
|
"""Validate and return space-separated WebDAV usernames (no @groups)."""
|
|
users = []
|
|
for u in re.split(r'[\s,]+', valid_users):
|
|
if not u:
|
|
continue
|
|
if not VALID_NAME.match(u):
|
|
raise ValueError(f'invalid WebDAV username: {u!r} (no groups)')
|
|
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, 'read_only': False,
|
|
'force_user': '', 'force_group': '', 'recycle_bin': False}
|
|
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 in ('writable', 'writeable'):
|
|
current['read_only'] = val.lower() in ('no', 'false', '0')
|
|
elif key == 'read_only':
|
|
current['read_only'] = val.lower() in ('yes', 'true', '1')
|
|
elif key == 'force_user':
|
|
current['force_user'] = val
|
|
elif key == 'force_group':
|
|
current['force_group'] = val
|
|
elif key == 'vfs_objects' and 'recycle' in val:
|
|
current['recycle_bin'] = True
|
|
|
|
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)
|
|
|
|
|
|
# ===== WEBDAV =====
|
|
|
|
WEBDAV_CONF = '/etc/apache2/conf-available/micronas-webdav.conf'
|
|
WEBDAV_PASSWD = '/etc/micronas/webdav.passwd'
|
|
_WEBDAV_MARK = 'MICRONAS'
|
|
|
|
|
|
def _parse_webdav_conf():
|
|
shares = []
|
|
try:
|
|
content = Path(WEBDAV_CONF).read_text()
|
|
except FileNotFoundError:
|
|
return shares
|
|
pattern = re.compile(
|
|
r'# BEGIN ' + _WEBDAV_MARK + r' (\S+)\n(.*?)# END ' + _WEBDAV_MARK + r' \1',
|
|
re.DOTALL,
|
|
)
|
|
for m in pattern.finditer(content):
|
|
name = m.group(1)
|
|
block = m.group(2)
|
|
alias_m = re.search(r'^Alias\s+\S+\s+(\S+)', block, re.M)
|
|
users_m = re.search(r'Require user (.+)$', block, re.M)
|
|
shares.append({
|
|
'name': name,
|
|
'path': alias_m.group(1).strip() if alias_m else '',
|
|
'users': users_m.group(1).strip() if users_m else '',
|
|
})
|
|
return shares
|
|
|
|
|
|
def _ensure_conf_dir():
|
|
"""Create /etc/micronas with correct ownership so Apache can traverse it."""
|
|
os.makedirs(_CONF_DIR, mode=0o750, exist_ok=True)
|
|
subprocess.run(['chown', 'root:www-data', _CONF_DIR], capture_output=True)
|
|
|
|
|
|
def _write_webdav_conf(shares):
|
|
_ensure_conf_dir()
|
|
os.makedirs('/var/lib/apache2/webdav', exist_ok=True)
|
|
subprocess.run(['chown', 'www-data:www-data', '/var/lib/apache2/webdav'], capture_output=True)
|
|
lines = ['# Managed by MicroNAS — do not edit manually\n\n',
|
|
'DavLockDB /var/lib/apache2/webdav/DavLock\n\n']
|
|
for s in shares:
|
|
require = f'Require user {s["users"]}' if s.get('users') else 'Require valid-user'
|
|
lines += [
|
|
f'# BEGIN {_WEBDAV_MARK} {s["name"]}\n',
|
|
f'Alias /dav/{s["name"]} {s["path"]}\n',
|
|
f'<Directory {s["path"]}>\n',
|
|
' Options Indexes\n',
|
|
' AllowOverride None\n',
|
|
' Require all granted\n',
|
|
'</Directory>\n',
|
|
f'<Location /dav/{s["name"]}>\n',
|
|
' DAV On\n',
|
|
' AuthType Basic\n',
|
|
' AuthName "MicroNAS WebDAV"\n',
|
|
f' AuthUserFile {WEBDAV_PASSWD}\n',
|
|
f' {require}\n',
|
|
'</Location>\n',
|
|
f'# END {_WEBDAV_MARK} {s["name"]}\n\n',
|
|
]
|
|
with open(WEBDAV_CONF, 'w') as f:
|
|
f.writelines(lines)
|
|
# Ensure passwd file exists and is readable by www-data (Apache) but not world-readable
|
|
if not os.path.exists(WEBDAV_PASSWD):
|
|
Path(WEBDAV_PASSWD).touch(mode=0o640)
|
|
subprocess.run(['chown', 'root:www-data', WEBDAV_PASSWD], capture_output=True)
|
|
subprocess.run(['a2enconf', 'micronas-webdav'], capture_output=True)
|
|
subprocess.run(['systemctl', 'reload', 'apache2'], capture_output=True)
|
|
|
|
|
|
def _set_webdav_password(name, password):
|
|
"""Add or update a user in the WebDAV htpasswd file. Silent if htpasswd not installed."""
|
|
flag = '-ci' if not os.path.exists(WEBDAV_PASSWD) else '-i'
|
|
try:
|
|
subprocess.run(['htpasswd', flag, WEBDAV_PASSWD, name],
|
|
input=password, capture_output=True, text=True)
|
|
except FileNotFoundError:
|
|
pass # apache2-utils not installed — skip
|
|
|
|
|
|
def _remove_webdav_user(name):
|
|
if os.path.exists(WEBDAV_PASSWD):
|
|
try:
|
|
subprocess.run(['htpasswd', '-D', WEBDAV_PASSWD, name], capture_output=True)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
@app.route('/api/share/webdav', methods=['POST'])
|
|
@require_auth
|
|
def create_webdav():
|
|
data = request.json or {}
|
|
name = data.get('name', '').strip()
|
|
path = data.get('path', '')
|
|
users = data.get('users', '')
|
|
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid share name'}), 400
|
|
try:
|
|
safe_path = validate_path(path)
|
|
clean_users = _validate_webdav_users(users) if users else ''
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
shares = _parse_webdav_conf()
|
|
if any(s['name'] == name for s in shares):
|
|
return jsonify({'error': f'share {name!r} already exists'}), 400
|
|
shares.append({'name': name, 'path': safe_path, 'users': clean_users})
|
|
_write_webdav_conf(shares)
|
|
return jsonify({'status': 'created'})
|
|
|
|
|
|
@app.route('/api/share/webdav/<name>', methods=['PATCH'])
|
|
@require_auth
|
|
def edit_webdav(name):
|
|
data = request.json or {}
|
|
new_name = data.get('name', name).strip()
|
|
path = data.get('path', '')
|
|
users = data.get('users', '')
|
|
|
|
if not VALID_NAME.match(new_name):
|
|
return jsonify({'error': 'invalid share name'}), 400
|
|
try:
|
|
safe_path = validate_path(path)
|
|
clean_users = _validate_webdav_users(users) if users else ''
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
shares = _parse_webdav_conf()
|
|
for s in shares:
|
|
if s['name'] == name:
|
|
s['name'] = new_name
|
|
s['path'] = safe_path
|
|
s['users'] = clean_users
|
|
break
|
|
else:
|
|
return jsonify({'error': f'share {name!r} not found'}), 404
|
|
_write_webdav_conf(shares)
|
|
return jsonify({'status': 'updated'})
|
|
|
|
|
|
@app.route('/api/share/webdav/<name>', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_webdav(name):
|
|
if not VALID_NAME.match(name):
|
|
return jsonify({'error': 'invalid share name'}), 400
|
|
shares = _parse_webdav_conf()
|
|
filtered = [s for s in shares if s['name'] != name]
|
|
if len(filtered) == len(shares):
|
|
return jsonify({'error': f'share {name!r} not found'}), 404
|
|
_write_webdav_conf(filtered)
|
|
return jsonify({'status': 'deleted'})
|
|
|
|
|
|
# ===== AUTH ROUTES =====
|
|
|
|
@app.route('/setup', methods=['GET', 'POST'])
|
|
def setup():
|
|
if _is_configured():
|
|
return redirect(url_for('index'))
|
|
error = None
|
|
username = 'admin'
|
|
if request.method == 'POST':
|
|
username = request.form.get('username', '').strip()
|
|
password = request.form.get('password', '')
|
|
confirm = request.form.get('confirm', '')
|
|
if not username or not VALID_NAME.match(username):
|
|
error = 'Invalid username (letters, digits, _ - only).'
|
|
elif len(password) < 8:
|
|
error = 'Password must be at least 8 characters.'
|
|
elif password != confirm:
|
|
error = 'Passwords do not match.'
|
|
else:
|
|
_save_credentials(username, password)
|
|
return redirect(url_for('login'))
|
|
return render_template('setup.html', error=error, username=username)
|
|
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if not _is_configured():
|
|
return redirect(url_for('setup'))
|
|
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 username == _creds['user'] and _verify_password(password):
|
|
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']))
|
|
webdav = _parse_webdav_conf()
|
|
for s in webdav:
|
|
s.update(_path_stat(s['path']))
|
|
return jsonify({'smb': smb, 'nfs': nfs, 'webdav': webdav})
|
|
|
|
|
|
# ===== 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)
|
|
read_only = data.get('readOnly', False)
|
|
recycle_bin = data.get('recycleBin', 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',
|
|
f' {"read only" if read_only else "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')
|
|
if recycle_bin:
|
|
lines.append(' vfs objects = recycle\n')
|
|
lines.append(' recycle:repository = .recycle\n')
|
|
lines.append(' recycle:keeptree = yes\n')
|
|
lines.append(' recycle:versions = 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'}
|
|
_USER_WHITELIST = {'www-data'} # system accounts useful as folder owners
|
|
|
|
|
|
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:
|
|
if name not in _USER_WHITELIST:
|
|
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
|
|
_set_webdav_password(name, password)
|
|
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)
|
|
_remove_webdav_user(name)
|
|
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/admin/credentials', methods=['POST'])
|
|
@require_auth
|
|
def change_admin_credentials():
|
|
data = request.json or {}
|
|
current_pw = data.get('currentPassword', '')
|
|
new_user = data.get('username', '').strip()
|
|
new_pw = data.get('password', '')
|
|
|
|
if not _verify_password(current_pw):
|
|
return jsonify({'error': 'Current password is incorrect.'}), 403
|
|
if new_user and not VALID_NAME.match(new_user):
|
|
return jsonify({'error': 'Invalid username.'}), 400
|
|
if new_pw and len(new_pw) < 8:
|
|
return jsonify({'error': 'Password must be at least 8 characters.'}), 400
|
|
|
|
user = new_user or _creds['user']
|
|
pw = new_pw or current_pw
|
|
_save_credentials(user, pw)
|
|
session.clear()
|
|
return jsonify({'status': 'updated'})
|
|
|
|
|
|
|
|
@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
|
|
_set_webdav_password(name, password)
|
|
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'),
|
|
'apache2': is_active('apache2'),
|
|
}
|
|
|
|
|
|
def _server_ips():
|
|
"""Return list of {iface, ip} for all non-loopback IPv4 addresses."""
|
|
try:
|
|
r = subprocess.run(['ip', '-4', '-o', 'addr', 'show'], capture_output=True, text=True)
|
|
result = []
|
|
for line in r.stdout.splitlines():
|
|
# Format: <idx>: <iface> inet <ip/prefix> ...
|
|
m = re.match(r'\d+:\s+(\S+)\s+inet\s+([\d.]+)', line)
|
|
if m and m.group(1) != 'lo':
|
|
result.append({'iface': m.group(1), 'ip': m.group(2)})
|
|
return result
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _active_connections():
|
|
"""Parse smbstatus output to return list of active SMB sessions."""
|
|
try:
|
|
r = subprocess.run(['smbstatus', '-b'], capture_output=True, text=True, timeout=5)
|
|
sessions = []
|
|
in_section = False
|
|
for line in r.stdout.splitlines():
|
|
if re.match(r'-{10,}', line):
|
|
in_section = True
|
|
continue
|
|
if not in_section or not line.strip():
|
|
continue
|
|
# "PID Username Group Machine Protocol Encryption Signing"
|
|
# Skip the header row
|
|
if line.strip().startswith('PID'):
|
|
continue
|
|
parts = line.split()
|
|
if len(parts) >= 4:
|
|
sessions.append({'user': parts[1], 'machine': parts[3]})
|
|
return sessions
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
@app.route('/api/service/<name>/restart', methods=['POST'])
|
|
@require_auth
|
|
def restart_service(name):
|
|
service_map = {'smbd': 'smbd', 'nfsd': 'nfs-server', 'apache2': 'apache2'}
|
|
if name not in service_map:
|
|
return jsonify({'error': 'unknown service'}), 400
|
|
r = subprocess.run(['systemctl', 'restart', service_map[name]], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return jsonify({'error': r.stderr.strip() or 'restart failed'}), 500
|
|
return jsonify({'status': 'restarted'})
|
|
|
|
|
|
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,
|
|
})
|
|
for s in _parse_webdav_conf():
|
|
result.append({
|
|
'type': 'webdav', 'name': s['name'], 'path': s['path'],
|
|
'access': s['users'] or 'all', 'enabled': True,
|
|
'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(),
|
|
'ips': _server_ips(),
|
|
'connections': _active_connections(),
|
|
})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000, debug=False)
|