MicroNAS/app.py

import os
import re
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 _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 =====

def _parse_exports():
    """Parse /etc/exports preserving blanks, comments, and disabled (commented-out) exports."""
    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('#'):
                    inner = stripped[1:].strip()
                    if inner.startswith('/'):
                        parts = inner.split(None, 1)
                        entries.append({'kind': 'export', 'enabled': False,
                                        'path': parts[0], 'rest': parts[1] if len(parts) > 1 else '',
                                        'raw': raw})
                    else:
                        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 '',
                                'raw': raw})
    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 '#'
                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():
    return jsonify({'smb': parse_smb(), 'nfs': get_nfs_shares()})


# ===== 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'})


# ===== QUOTA =====

@app.route('/api/quota', methods=['POST'])
@require_auth
def set_quota():
    data = request.json or {}
    user = data.get('user', '')
    quota_type = data.get('type', 'smb')
    raw_quota = data.get('quota', 0)
    path = data.get('path', '/srv/samba')

    if not VALID_NAME.match(user):
        return jsonify({'error': 'invalid username'}), 400
    try:
        quota_gb = int(raw_quota)
        if quota_gb <= 0:
            raise ValueError()
    except (ValueError, TypeError):
        return jsonify({'error': 'quota must be a positive integer (GB)'}), 400

    if quota_type == 'smb':
        mount = '/srv/samba'
    else:
        try:
            mount = validate_path(path)
        except ValueError as e:
            return jsonify({'error': str(e)}), 400

    soft_kb = str(quota_gb * 1024 * 1024)
    hard_kb = str((quota_gb + 1) * 1024 * 1024)
    subprocess.run(['setquota', '-u', user, soft_kb, hard_kb, '0', '0', mount], check=True)
    return jsonify({'status': 'ok'})


# ===== 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


def _quota_usage():
    try:
        r = subprocess.run(['repquota', '-u', '/srv'], capture_output=True, text=True, timeout=10)
    except FileNotFoundError:
        return []

    quotas = []
    past_sep = False
    for line in r.stdout.splitlines():
        if line.startswith('---'):
            past_sep = True
            continue
        if not past_sep:
            continue
        parts = line.split()
        if len(parts) < 5 or not re.match(r'^[+-]{2}$', parts[1]):
            continue
        try:
            used_kb, soft_kb, hard_kb = int(parts[2]), int(parts[3]), int(parts[4])
        except (ValueError, IndexError):
            continue
        if soft_kb == 0 and hard_kb == 0 and used_kb == 0:
            continue
        quotas.append({
            'user': parts[0],
            'used_gb': round(used_kb / 1024**2, 2),
            'soft_gb': round(soft_kb / 1024**2, 2),
            'hard_gb': round(hard_kb / 1024**2, 2),
            'percent': round(used_kb / soft_kb * 100, 1) if soft_kb else 0,
        })
    return quotas


@app.route('/api/dashboard', methods=['GET'])
@require_auth
def get_dashboard():
    return jsonify({
        'status': _daemon_status(),
        'disk': _disk_usage(),
        'shares': _shares_with_usage(),
        'quotas': _quota_usage(),
    })


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)