app.py

import os
import logging
import copy
import hmac
import secrets
import threading
import time
import re
from datetime import datetime, timezone
from urllib.parse import urlparse
from typing import Any, Dict
from flask import Flask, render_template, request, redirect, url_for, session, jsonify, abort
from werkzeug.security import check_password_hash, generate_password_hash

# Robust import for config and utils
try:
    import config
    from utils import file_handler, stream_handler, thumbnail, network, subtitle_utils, runtime_checks
    from services import persistence
except ImportError:
    from StreamHive import config  # type: ignore[no-redef]
    from StreamHive.utils import (  # type: ignore[no-redef]
        file_handler,
        stream_handler,
        thumbnail,
        network,
        subtitle_utils,
        runtime_checks,
    )
    from StreamHive.services import persistence  # type: ignore[no-redef]

app = Flask(__name__)
app.secret_key = config.SECRET_KEY
app.config.from_object(config)
app.config["SESSION_COOKIE_HTTPONLY"] = True
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
app.config["SESSION_COOKIE_SECURE"] = bool(getattr(config, "SESSION_COOKIE_SECURE", False))

logging.basicConfig(
    level=os.getenv("STREAMHIVE_LOG_LEVEL", "INFO").upper(),
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger("streamhive")
if not os.getenv("STREAMHIVE_SECRET_KEY"):
    logger.warning("STREAMHIVE_SECRET_KEY is not set. Generated ephemeral secret key is in use.")

_login_attempts: Dict[str, Dict[str, Any]] = {}
_login_attempts_lock = threading.Lock()
_users_lock = threading.Lock()
_health_cache_lock = threading.Lock()
_health_cache_payload = None
_health_cache_expires_at = 0.0
USERS_FILE_PATH = os.path.join(getattr(config, "BASE_DIR", os.path.dirname(os.path.abspath(__file__))), "users.json")
CONFIG_FILE_PATH = os.path.join(getattr(config, "BASE_DIR", os.path.dirname(os.path.abspath(__file__))), "config.json")
ROLE_ADMIN = "admin"
ROLE_USER = "user"
ROLE_GUEST = "guest"
USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_.-]{3,32}$")


def _coerce_bool(value, default=False):
    """
    Strict boolean coercion for API payloads/config values.
    """
    if isinstance(value, bool):
        return value
    if value is None:
        return bool(default)
    if isinstance(value, (int, float)):
        return value != 0
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in {"1", "true", "yes", "on", "y"}:
            return True
        if normalized in {"0", "false", "no", "off", "n", ""}:
            return False
    return bool(default)


def _apply_runtime_safety():
    """
    Normalize shared folders and ensure runtime directories are ready.
    """
    normalized_folders, invalid_folders = runtime_checks.normalize_shared_folders(config.SHARED_FOLDERS)
    if invalid_folders:
        logger.warning("Ignoring invalid shared folders: %s", invalid_folders)
    config.SHARED_FOLDERS[:] = normalized_folders

    if not runtime_checks.ensure_thumbnail_cache(config.THUMBNAIL_CACHE_DIR):
        logger.warning("Thumbnail cache directory is not writable: %s", config.THUMBNAIL_CACHE_DIR)


def _invalidate_health_cache():
    global _health_cache_payload, _health_cache_expires_at
    with _health_cache_lock:
        _health_cache_payload = None
        _health_cache_expires_at = 0.0


def _build_health_payload(force_refresh=False):
    global _health_cache_payload, _health_cache_expires_at
    now = time.monotonic()
    ttl = max(0, int(getattr(config, "HEALTH_CACHE_TTL_SEC", 5)))

    with _health_cache_lock:
        if ttl > 0 and not force_refresh and _health_cache_payload is not None and _health_cache_expires_at > now:
            return copy.deepcopy(_health_cache_payload)

    _apply_runtime_safety()

    try:
        ip = network.get_local_ip(cache_ttl_sec=getattr(config, "IP_CACHE_TTL_SEC", 30))
    except Exception:
        ip = "127.0.0.1"

    snapshot = runtime_checks.build_health_snapshot(
        shared_folders=config.SHARED_FOLDERS,
        ffmpeg_binary=config.FFMPEG_BINARY,
        thumbnail_cache_dir=config.THUMBNAIL_CACHE_DIR,
        thumbnail_enabled=config.ENABLE_THUMBNAIL,
        lightweight=True,
    )

    payload = {
        "timestamp_utc": datetime.now(timezone.utc).isoformat(),
        "ip": ip,
        "port": config.PORT,
        "url": f"http://{ip}:{config.PORT}",
        "status": snapshot["status"],
        "issues": snapshot["issues"],
        "shared_folders": {
            "configured": snapshot["shared_folders_total"],
            "accessible": snapshot["shared_folders_accessible"],
        },
        "ffmpeg": {
            "available": snapshot["ffmpeg_available"],
        },
        "thumbnail": {
            "enabled": snapshot["thumbnail_enabled"],
            "cache_ready": snapshot["thumbnail_cache_ready"],
        },
    }

    with _health_cache_lock:
        _health_cache_payload = payload
        _health_cache_expires_at = now + ttl

    return copy.deepcopy(payload)


_apply_runtime_safety()


def _load_users_snapshot_cached(users_path, users_mtime_ns):
    return persistence.load_users_snapshot_cached(
        users_path,
        users_mtime_ns,
        role_admin=ROLE_ADMIN,
        role_user=ROLE_USER,
    )


def _get_users_snapshot():
    return persistence.get_users_snapshot(
        USERS_FILE_PATH,
        role_admin=ROLE_ADMIN,
        role_user=ROLE_USER,
    )


def _snapshot_row_to_user(row):
    return persistence.snapshot_row_to_user(row)


def _load_users_store():
    """
    Loads users from users.json. Returns empty list when file is missing/invalid.
    """
    return persistence.load_users_store(
        USERS_FILE_PATH,
        role_admin=ROLE_ADMIN,
        role_user=ROLE_USER,
    )


def _write_users_store(users):
    """
    Persists users atomically to users.json.
    """
    persistence.write_users_store(USERS_FILE_PATH, users)


def _is_multiuser_auth_enabled():
    """
    Multi-user auth is active when users.json exists and has at least one account.
    """
    if os.getenv("STREAMHIVE_MULTIUSER_AUTH", "1") != "1":
        return False
    return len(_get_users_snapshot()) > 0


def _find_user(username):
    target = str(username or "").strip().lower()
    if not target:
        return None
    for row in _get_users_snapshot():
        if row[0] == target:
            return _snapshot_row_to_user(row)
    return None


def _list_admin_users():
    users = []
    for row in _get_users_snapshot():
        user = _snapshot_row_to_user(row)
        if user.get("role") == ROLE_ADMIN and user.get("active"):
            users.append(user)
    return users


def _serialize_user_public(user):
    return {
        "username": user.get("username"),
        "role": user.get("role"),
        "active": bool(user.get("active")),
        "created_at": user.get("created_at", ""),
        "last_login_at": user.get("last_login_at", ""),
    }


def _save_runtime_config():
    """
    Saves shared folders + port back to config.json for launcher/web consistency.
    """
    persistence.save_runtime_config(
        CONFIG_FILE_PATH,
        shared_folders=config.SHARED_FOLDERS,
        port=config.PORT,
    )


def _current_role():
    return str(session.get("role", ROLE_USER)).lower()


def _is_admin_session():
    return session.get("logged_in") and _current_role() == ROLE_ADMIN


def _is_guest_session():
    return session.get("logged_in") and _current_role() == ROLE_GUEST


def _require_csrf_json():
    """
    For JSON mutating routes: requires valid CSRF token in X-CSRF-Token header.
    """
    expected = session.get("csrf_token", "")
    provided = request.headers.get("X-CSRF-Token", "")
    if not expected or not hmac.compare_digest(str(provided), str(expected)):
        abort(400, description="Invalid CSRF token.")


def _admin_required_json():
    if not _is_admin_session():
        abort(403)


@app.context_processor
def inject_view_context():
    username = session.get("username", "")
    role = _current_role()
    return {
        "current_username": username,
        "current_role": role,
        "is_admin": bool(session.get("logged_in")) and role == ROLE_ADMIN,
        "is_guest": bool(session.get("logged_in")) and role == ROLE_GUEST,
        "csrf_token": _ensure_csrf_token(),
    }


def _wants_json_response():
    """
    Returns True when request prefers JSON or targets API routes.
    """
    if request.path.startswith("/api/"):
        return True

    best = request.accept_mimetypes.best_match(["application/json", "text/html"])
    return best == "application/json" and request.accept_mimetypes[best] > request.accept_mimetypes["text/html"]


def _get_client_ip():
    """
    Returns best-effort client IP for access control and rate limiting.
    """
    forwarded = request.headers.get("X-Forwarded-For", "").strip()
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.remote_addr or "unknown"


def _is_local_request():
    """
    True when request originates from localhost loopback.
    """
    return _get_client_ip() in {"127.0.0.1", "::1", "localhost"}


def _is_safe_next_url(target):
    """
    Prevent open redirect by allowing only same-host or relative URLs.
    """
    if not target:
        return False
    base = urlparse(request.host_url)
    test = urlparse(target)
    if test.scheme and test.scheme not in {"http", "https"}:
        return False
    if test.netloc and test.netloc != base.netloc:
        return False
    return test.path.startswith("/")


def _ensure_csrf_token():
    """
    Ensures CSRF token exists in session.
    """
    token = session.get("csrf_token")
    if not token:
        token = secrets.token_urlsafe(32)
        session["csrf_token"] = token
    return token


def _is_rate_limited(client_ip):
    """
    Checks login throttling state for an IP.
    """
    now = time.time()
    max_attempts = max(1, int(getattr(config, "LOGIN_RATE_LIMIT_MAX_ATTEMPTS", 5)))
    window_sec = max(1, int(getattr(config, "LOGIN_RATE_LIMIT_WINDOW_SEC", 300)))
    block_sec = max(1, int(getattr(config, "LOGIN_RATE_LIMIT_BLOCK_SEC", 900)))

    with _login_attempts_lock:
        entry = _login_attempts.get(client_ip)
        if not entry:
            return False, 0

        blocked_until = entry.get("blocked_until", 0)
        if blocked_until > now:
            wait_for = int(blocked_until - now)
            return True, wait_for

        attempts = [stamp for stamp in entry.get("attempts", []) if now - stamp <= window_sec]
        entry["attempts"] = attempts
        if len(attempts) >= max_attempts:
            entry["blocked_until"] = now + block_sec
            wait_for = int(block_sec)
            return True, wait_for

        return False, 0


def _record_failed_login(client_ip):
    """
    Tracks failed login attempts per IP.
    """
    now = time.time()
    window_sec = max(1, int(getattr(config, "LOGIN_RATE_LIMIT_WINDOW_SEC", 300)))

    with _login_attempts_lock:
        entry = _login_attempts.setdefault(client_ip, {"attempts": [], "blocked_until": 0})
        attempts = [stamp for stamp in entry["attempts"] if now - stamp <= window_sec]
        attempts.append(now)
        entry["attempts"] = attempts


def _clear_login_failures(client_ip):
    """
    Clears login throttling state after successful authentication.
    """
    with _login_attempts_lock:
        _login_attempts.pop(client_ip, None)


def _safe_next_url():
    next_url = request.args.get("next", "")
    if _is_safe_next_url(next_url):
        return next_url
    return ""


def _redirect_after_login(next_url):
    if _is_safe_next_url(next_url):
        return redirect(next_url)
    return redirect(url_for("media.index"))


def _redirect_to_login():
    return redirect(url_for("auth.login"))


def _set_authenticated_session(username, role):
    session["logged_in"] = True
    session["username"] = username
    session["role"] = role
    session["csrf_token"] = secrets.token_urlsafe(32)


def _render_login_page(
    error=None,
    status_code=200,
    next_url="",
    csrf_token=None,
    uses_multiuser=None,
    allow_guest_mode=None,
):
    if csrf_token is None:
        csrf_token = _ensure_csrf_token()
    if uses_multiuser is None:
        uses_multiuser = _is_multiuser_auth_enabled()
    if allow_guest_mode is None:
        allow_guest_mode = bool(getattr(config, "ALLOW_GUEST_LAN", False))

    payload = {
        "csrf_token": csrf_token,
        "uses_multiuser": uses_multiuser,
        "allow_guest_mode": allow_guest_mode,
        "next_url": next_url,
    }
    if error:
        payload["error"] = error

    page = render_template("login.html", **payload)
    if status_code == 200:
        return page
    return page, status_code


def _validate_multiuser_password(user, password):
    if not user or not user.get("active"):
        return False, False

    if check_password_hash(user.get("password_hash", ""), password):
        return True, False

    admin_username = (getattr(config, "ADMIN_USERNAME", "admin") or "admin").strip().lower()
    if (
        user.get("username") == admin_username
        and config.PASSWORD
        and hmac.compare_digest(password, str(config.PASSWORD))
    ):
        return True, True

    return False, False


def _mark_user_login(username, password, used_env_admin_password):
    with _users_lock:
        users = _load_users_store()
        for item in users:
            if item.get("username") == username:
                item["last_login_at"] = datetime.now(timezone.utc).isoformat()
                if used_env_admin_password:
                    item["password_hash"] = generate_password_hash(password)
                break
        _write_users_store(users)


def _build_root_entries():
    roots = []
    for index, folder in enumerate(config.SHARED_FOLDERS):
        name = os.path.basename(folder) or folder
        roots.append(
            {
                "name": name,
                "path": str(index),
                "is_dir": True,
                "type": "folder",
                "size_human": "",
                "modified_date": "",
            }
        )
    return roots


def _build_breadcrumbs(subpath):
    breadcrumbs = []
    try:
        parts = subpath.strip("/").split("/")
        root_index = int(parts[0])
        root_path = config.SHARED_FOLDERS[root_index]
        root_name = os.path.basename(root_path) or root_path
        breadcrumbs.append({"name": root_name, "url": url_for("media.browse", subpath=str(root_index))})

        current_link = str(root_index)
        for part in parts[1:]:
            current_link += f"/{part}"
            breadcrumbs.append({"name": part, "url": url_for("media.browse", subpath=current_link)})
    except (ValueError, IndexError):
        return []
    return breadcrumbs


def _split_media_filepath(filepath):
    parts = filepath.strip("/").split("/", 1)
    root_token = parts[0]
    root_index = int(root_token)
    rel_path = parts[1] if len(parts) > 1 else ""
    parent_rel = os.path.dirname(rel_path)
    parent_subpath = root_token if not parent_rel else f"{root_token}/{parent_rel}"
    return root_token, root_index, parent_subpath


def _json_error(message, status_code):
    return jsonify({"status": "error", "message": message}), status_code


def _refresh_runtime_after_folder_update():
    _apply_runtime_safety()
    _save_runtime_config()
    _invalidate_health_cache()


def _require_multiuser_json():
    if _is_multiuser_auth_enabled():
        return None
    return _json_error("Multi-user auth is not initialized.", 400)


# Helper to resolve paths from URL subpath
def resolve_path(subpath):
    """
    Parses subpath "root_index/relative_path" to an absolute path.
    Security: Ensures the resolved path is within the shared folder.
    """
    try:
        parts = subpath.strip("/").split("/", 1)
        root_index = int(parts[0])
        rel_path = parts[1] if len(parts) > 1 else ""

        if root_index < 0 or root_index >= len(config.SHARED_FOLDERS):
            return None

        base_path = config.SHARED_FOLDERS[root_index]
        base_real = os.path.realpath(base_path)
        abs_path = os.path.realpath(os.path.join(base_real, rel_path))

        # Security check: Ensure path is within base_path
        if os.path.commonpath([base_real, abs_path]) != base_real:
            return None

        return abs_path
    except (ValueError, IndexError):
        return None


def _is_within_root(path_value, root_value):
    """
    Fast path prefix check that safely handles filesystem roots (/, C:\\).
    """
    path_normalized = os.path.normcase(os.path.abspath(path_value))
    root_normalized = os.path.normcase(os.path.abspath(root_value))

    if path_normalized == root_normalized:
        return True

    if root_normalized.endswith(os.sep):
        return path_normalized.startswith(root_normalized)

    return path_normalized.startswith(root_normalized + os.sep)


# Middleware for Password Protection
@app.before_request
def require_login():
    if request.method == "OPTIONS":
        return

    endpoint = request.endpoint or ""
    is_multiuser = _is_multiuser_auth_enabled()

    allowed_routes = {
        "auth.login",
        "auth.guest_login",
        "static",
        "system.favicon",
        "system.web_manifest",
        "system.service_worker",
    }
    if endpoint in allowed_routes:
        return

    if is_multiuser:
        if not session.get("logged_in"):
            return redirect(url_for("auth.login", next=request.path))
        return

    if not config.PASSWORD:
        # Secure-by-default: when no password is configured, only local requests
        # are allowed unless explicitly enabled through config.
        if not getattr(config, "ALLOW_GUEST_LAN", False) and not _is_local_request():
            return abort(403)
        return

    if not session.get("logged_in"):
        return redirect(url_for("auth.login", next=request.path))


def _build_blueprint_dependencies():
    return {
        "app": app,
        "config": config,
        "logger": logger,
        "network": network,
        "file_handler": file_handler,
        "stream_handler": stream_handler,
        "thumbnail": thumbnail,
        "subtitle_utils": subtitle_utils,
        "ROLE_ADMIN": ROLE_ADMIN,
        "ROLE_USER": ROLE_USER,
        "ROLE_GUEST": ROLE_GUEST,
        "USERNAME_PATTERN": USERNAME_PATTERN,
        "users_lock": _users_lock,
        "coerce_bool": _coerce_bool,
        "load_users_store": _load_users_store,
        "write_users_store": _write_users_store,
        "find_user": _find_user,
        "is_multiuser_auth_enabled": _is_multiuser_auth_enabled,
        "ensure_csrf_token": _ensure_csrf_token,
        "get_client_ip": _get_client_ip,
        "safe_next_url": _safe_next_url,
        "redirect_after_login": _redirect_after_login,
        "redirect_to_login": _redirect_to_login,
        "is_rate_limited": _is_rate_limited,
        "record_failed_login": _record_failed_login,
        "clear_login_failures": _clear_login_failures,
        "set_authenticated_session": _set_authenticated_session,
        "render_login_page": _render_login_page,
        "validate_multiuser_password": _validate_multiuser_password,
        "mark_user_login": _mark_user_login,
        "resolve_path": resolve_path,
        "is_within_root": _is_within_root,
        "is_guest_session": _is_guest_session,
        "build_root_entries": _build_root_entries,
        "build_breadcrumbs": _build_breadcrumbs,
        "split_media_filepath": _split_media_filepath,
        "build_health_payload": _build_health_payload,
        "serialize_user_public": _serialize_user_public,
        "admin_required_json": _admin_required_json,
        "require_csrf_json": _require_csrf_json,
        "json_error": _json_error,
        "refresh_runtime_after_folder_update": _refresh_runtime_after_folder_update,
        "require_multiuser_json": _require_multiuser_json,
    }


app.extensions["streamhive_deps"] = _build_blueprint_dependencies()

try:
    from routes.auth import bp as auth_bp
    from routes.media import bp as media_bp
    from routes.admin import bp as admin_bp
    from routes.system import bp as system_bp
except ImportError:
    from StreamHive.routes.auth import bp as auth_bp  # type: ignore[no-redef]
    from StreamHive.routes.media import bp as media_bp  # type: ignore[no-redef]
    from StreamHive.routes.admin import bp as admin_bp  # type: ignore[no-redef]
    from StreamHive.routes.system import bp as system_bp  # type: ignore[no-redef]


app.register_blueprint(auth_bp)
app.register_blueprint(media_bp)
app.register_blueprint(admin_bp)
app.register_blueprint(system_bp)


@app.after_request
def add_security_headers(response):
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
    response.headers.setdefault("X-Frame-Options", "DENY")
    response.headers.setdefault("Referrer-Policy", "same-origin")
    response.headers.setdefault(
        "Content-Security-Policy",
        "default-src 'self'; "
        "img-src 'self' data:; "
        "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdnjs.cloudflare.com; "
        "font-src 'self' https://fonts.gstatic.com https://cdnjs.cloudflare.com data:; "
        "script-src 'self' 'unsafe-inline'; "
        "connect-src 'self'; "
        "form-action 'self'; "
        "base-uri 'self'; "
        "frame-ancestors 'none'",
    )
    return response


@app.errorhandler(403)
def handle_forbidden(error):
    if _wants_json_response():
        return (
            jsonify(
                {
                    "status": "error",
                    "error": "forbidden",
                    "message": "Access denied.",
                }
            ),
            403,
        )
    return render_template("errors/403.html"), 403


@app.errorhandler(400)
def handle_bad_request(error):
    if _wants_json_response():
        return (
            jsonify(
                {
                    "status": "error",
                    "error": "bad_request",
                    "message": str(getattr(error, "description", "Invalid request.")) or "Invalid request.",
                }
            ),
            400,
        )
    return render_template("errors/400.html"), 400


@app.errorhandler(404)
def handle_not_found(error):
    if _wants_json_response():
        return (
            jsonify(
                {
                    "status": "error",
                    "error": "not_found",
                    "message": "Requested resource was not found.",
                }
            ),
            404,
        )
    return render_template("errors/404.html"), 404


@app.errorhandler(500)
def handle_internal_error(error):
    logger.exception("Unhandled server error: %s", error)
    if _wants_json_response():
        return (
            jsonify(
                {
                    "status": "error",
                    "error": "internal_server_error",
                    "message": "Unexpected server error occurred.",
                }
            ),
            500,
        )
    return render_template("errors/500.html"), 500


if __name__ == "__main__":
    # Standalone run — uses threaded mode for multi-user support.
    # For launcher.py (GUI), werkzeug make_server with threaded=True is used instead.
    _apply_runtime_safety()

    if not config.SHARED_FOLDERS:
        logger.warning("No shared folders configured. Run launcher.py to configure.")

    health = runtime_checks.build_health_snapshot(
        shared_folders=config.SHARED_FOLDERS,
        ffmpeg_binary=config.FFMPEG_BINARY,
        thumbnail_cache_dir=config.THUMBNAIL_CACHE_DIR,
        thumbnail_enabled=config.ENABLE_THUMBNAIL,
    )

    if not config.PASSWORD and getattr(config, "ALLOW_GUEST_LAN", False):
        logger.warning("Guest LAN mode is enabled without password. Set STREAMHIVE_PASSWORD for secure access.")
    elif not config.PASSWORD:
        logger.info("Guest LAN mode is disabled. Only localhost access is allowed without password.")

    bind_host = getattr(config, "HOST", "0.0.0.0")
    try:
        lan_ip = network.get_local_ip(cache_ttl_sec=getattr(config, "IP_CACHE_TTL_SEC", 30))
    except Exception:
        lan_ip = "127.0.0.1"
    logger.info("StreamHive running on http://%s:%s", bind_host, config.PORT)
    logger.info("LAN URL: http://%s:%s", lan_ip, config.PORT)
    logger.info("Threaded mode: ON | Debug: OFF | Max concurrent streams: unlimited")
    logger.info(
        "Health status: %s | Shared folders: %s/%s",
        health["status"],
        health["shared_folders_accessible"],
        health["shared_folders_total"],
    )
    app.run(
        host=bind_host,
        port=config.PORT,
        debug=False,
        threaded=True,
        use_reloader=False,
    )