MicroNAS/app.py

import os
import re
import shutil
import subprocess
import functools
from pathlib import Path

from flask import Flask, request, jsonify, send_file, Response

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', '')

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)
        auth = request.authorization
        if not auth or auth.username != _AUTH_USER or auth.password != _AUTH_PASS:
            return Response(
                'Unauthorized', 401,
                {'WWW-Authenticate': 'Basic realm="micronas"'},
            )
        return f(*args, **kwargs)
    return decorated


def validate_path(path_str):
    """Resolve path and ensure it falls under /srv."""
    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)


# ===== SMB =====

_SMB_SKIP = {'global', 'homes', 'printers', 'print$', 'ipc$'}


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': ''}
        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

    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 get_nfs_shares():
    shares = []
    try:
        with open('/etc/exports', 'r') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue
                parts = line.split()
                shares.append({
                    'path': parts[0],
                    'clients': ' '.join(parts[1:]) if len(parts) > 1 else '*',
                })
    except FileNotFoundError:
        pass
    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)


# ===== ROUTES =====

@app.route('/')
@require_auth
def index():
    return send_file(os.path.join(os.path.dirname(__file__), 'index.html'))


@app.route('/api/shares', methods=['GET'])
@require_auth
def get_all_shares():
    return jsonify({'smb': parse_smb(), 'nfs': get_nfs_shares()})


@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
    users = []
    for u in re.split(r'[\s,]+', valid_users):
        if not u:
            continue
        if not VALID_NAME.match(u):
            return jsonify({'error': f'invalid username: {u}'}), 400
        users.append(u)
    try:
        safe_path = validate_path(path)
    except ValueError as e:
        return jsonify({'error': str(e)}), 400

    add_smb_share(name, safe_path, ', '.join(users))
    return jsonify({'type': 'smb', 'status': 'created'})


@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/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

    # setquota takes kilobytes; soft = requested, hard = +1 GB buffer
    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[1], 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'],
            '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'],
            '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 = int(parts[2])
            soft_kb = int(parts[3])
            hard_kb = 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)