diff --git a/cache.py b/cache.py index adfd18a..7f1661d 100644 --- a/cache.py +++ b/cache.py @@ -506,6 +506,8 @@ def load_server_settings() -> dict[str, Any]: data.setdefault("users", {}) data.setdefault("user_agent_preset", "tivimate") data.setdefault("user_agent_custom", "") + data.setdefault("m3u_export_enabled", False) + data.setdefault("m3u_export_mode", "redirect") # "redirect" or "proxy" return data diff --git a/m3u_export.py b/m3u_export.py new file mode 100644 index 0000000..1d55e72 --- /dev/null +++ b/m3u_export.py @@ -0,0 +1,228 @@ +"""M3U playlist / XMLTV export for external IPTV player apps (TiviMate, Smarters, VLC). + +Exposes the live channels a neTV user can access as an Xtream-style playlist: + + {base_url}/get.php?username=U&password=P -> M3U playlist + {base_url}/xmltv.php?username=U&password=P -> XMLTV guide + {base_url}/live/U/P/{stream_id}.ts -> channel playback + +Playback has two modes (server setting ``m3u_export_mode``): + +- ``redirect``: 302 to the upstream URL. Zero server bandwidth, but the + provider URL (including provider credentials for Xtream sources) is + visible to the client. +- ``proxy``: stream bytes through this server. Hides the provider and + enforces per-user/per-source stream limits, at the cost of upstream + + downstream bandwidth per viewer. HLS (.m3u8) direct URLs are always + redirected since proxying would require rewriting segment URLs. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from datetime import UTC, datetime, timedelta +from xml.sax.saxutils import ( + escape as xml_escape, + quoteattr as xml_quoteattr, +) + +import hashlib +import hmac +import logging +import threading +import time +import urllib.parse + +from cache import get_sources +from ffmpeg_command import get_user_agent +from util import safe_urlopen + +import auth +import epg + + +log = logging.getLogger(__name__) + +# Verified-credential cache: PBKDF2 verification is ~100ms on small ARM +# boards, too slow to run on every channel zap. username -> (sha256, expiry). +_CRED_CACHE: dict[str, tuple[str, float]] = {} +_CRED_CACHE_TTL = 300 +_cred_lock = threading.Lock() + +# Active proxied streams: (username, source_id) -> count. Tracked separately +# from web transcode sessions (ffmpeg_session), so limits apply per system. +_active_streams: dict[tuple[str, str], int] = {} +_streams_lock = threading.Lock() + +XMLTV_WINDOW_HOURS = 48 +_PROXY_CHUNK_SIZE = 64 * 1024 +_STRIP_EXTS = (".ts", ".m3u8") + + +def verify_credentials(username: str, password: str) -> bool: + """Verify username/password, caching successes to avoid repeated PBKDF2.""" + digest = hashlib.sha256(password.encode()).hexdigest() + now = time.time() + with _cred_lock: + cached = _CRED_CACHE.get(username) + if cached and cached[1] > now and hmac.compare_digest(cached[0], digest): + return True + if not auth.verify_password(username, password): + return False + with _cred_lock: + _CRED_CACHE[username] = (digest, now + _CRED_CACHE_TTL) + # Bounded: one entry per user, plus purge anything expired + for u in [u for u, (_, exp) in _CRED_CACHE.items() if exp <= now]: + del _CRED_CACHE[u] + return True + + +def clear_credential_cache(username: str | None = None) -> None: + """Drop cached credentials (call after password change/user delete).""" + with _cred_lock: + if username is None: + _CRED_CACHE.clear() + else: + _CRED_CACHE.pop(username, None) + + +def stream_allowed(stream: dict, unavailable_groups: set[str]) -> bool: + """Same rule as the guide: blocked if ANY of its categories is unavailable.""" + cat_ids = stream.get("category_ids") or [] + return not any(f"cat:{c}" in unavailable_groups for c in cat_ids) + + +def allowed_live_streams(streams: list[dict], username: str) -> list[dict]: + """Filter live streams down to what this user may access.""" + limits = auth.get_user_limits(username) + unavailable = set(limits.get("unavailable_groups", [])) + if not unavailable: + return streams + return [s for s in streams if stream_allowed(s, unavailable)] + + +def upstream_url(stream: dict) -> str: + """Resolve a live stream dict to its upstream URL (MPEG-TS for Xtream).""" + if stream.get("direct_url"): + return stream["direct_url"] + if stream.get("source_type") == "xtream": + user = urllib.parse.quote(stream["source_username"], safe="") + pwd = urllib.parse.quote(stream["source_password"], safe="") + stream_id = str(stream["stream_id"]) + orig_id = stream_id.split("_")[-1] if "_" in stream_id else stream_id + return f"{stream['source_url']}/live/{user}/{pwd}/{orig_id}.ts" + return "" + + +def strip_stream_ext(stream_spec: str) -> str: + """Strip a trailing .ts/.m3u8 that players append to playback URLs.""" + for ext in _STRIP_EXTS: + if stream_spec.endswith(ext): + return stream_spec[: -len(ext)] + return stream_spec + + +def build_playlist( + categories: list[dict], + streams: list[dict], + base_url: str, + username: str, + password: str, +) -> str: + """Build an m3u_plus playlist pointing back at this server's /live/ URLs.""" + cat_names = {str(c.get("category_id")): c.get("category_name", "") for c in categories} + user = urllib.parse.quote(username, safe="") + pwd = urllib.parse.quote(password, safe="") + lines = ["#EXTM3U"] + for s in streams: + name = s.get("name", "Unknown") + tvg_id = s.get("epg_channel_id") or "" + logo = s.get("stream_icon") or "" + group = next( + (cat_names[str(c)] for c in (s.get("category_ids") or []) if str(c) in cat_names), + "", + ) + attrs = f'tvg-id="{tvg_id}" tvg-name="{name}" tvg-logo="{logo}" group-title="{group}"' + lines.append(f"#EXTINF:-1 {attrs},{name}") + lines.append(f"{base_url}/live/{user}/{pwd}/{urllib.parse.quote(str(s['stream_id']))}.ts") + return "\n".join(lines) + "\n" + + +def build_xmltv(streams: list[dict], hours: int = XMLTV_WINDOW_HOURS) -> str: + """Build an XMLTV guide for the given streams from the local EPG database.""" + epg_ids = sorted({s.get("epg_channel_id") or "" for s in streams} - {""}) + now = datetime.now(UTC) + programs_map = epg.get_programs_batch(epg_ids, now - timedelta(hours=2), now + timedelta(hours=hours)) + icons_map = epg.get_icons_batch(epg_ids) + + out = ['', ''] + seen: set[str] = set() + for s in streams: + epg_id = s.get("epg_channel_id") or "" + if not epg_id or epg_id in seen: + continue + seen.add(epg_id) + out.append(f" ") + out.append(f" {xml_escape(s.get('name', ''))}") + icon = s.get("stream_icon") or icons_map.get(epg_id, "") + if icon: + out.append(f" ") + out.append(" ") + for epg_id in epg_ids: + for p in programs_map.get(epg_id, []): + start = p.start.strftime("%Y%m%d%H%M%S %z") + stop = p.stop.strftime("%Y%m%d%H%M%S %z") + out.append( + f' ' + ) + out.append(f" {xml_escape(p.title)}") + if p.desc: + out.append(f" {xml_escape(p.desc)}") + out.append(" ") + out.append("") + return "\n".join(out) + "\n" + + +def _source_max_streams(source_id: str) -> int: + source = next((s for s in get_sources() if s.id == source_id), None) + return source.max_streams if source else 0 + + +def try_acquire_stream(username: str, source_id: str) -> bool: + """Reserve a proxied stream slot. Returns False if a limit is hit (0 = unlimited).""" + user_max = auth.get_user_limits(username).get("max_streams_per_source", {}).get(source_id, 0) + source_max = _source_max_streams(source_id) + with _streams_lock: + user_count = _active_streams.get((username, source_id), 0) + source_count = sum(n for (_, sid), n in _active_streams.items() if sid == source_id) + if user_max and user_count >= user_max: + log.info("Proxy stream denied: user=%s source=%s at user limit", username, source_id) + return False + if source_max and source_count >= source_max: + log.info("Proxy stream denied: user=%s source=%s at source limit", username, source_id) + return False + _active_streams[(username, source_id)] = user_count + 1 + return True + + +def release_stream(username: str, source_id: str) -> None: + """Release a slot reserved by try_acquire_stream.""" + with _streams_lock: + key = (username, source_id) + count = _active_streams.get(key, 0) + if count <= 1: + _active_streams.pop(key, None) + else: + _active_streams[key] = count - 1 + + +def iter_proxy_stream(url: str, username: str, source_id: str, timeout: int = 15) -> Iterator[bytes]: + """Yield upstream bytes; caller must have acquired a slot, released here.""" + try: + with safe_urlopen(url, timeout=timeout, user_agent=get_user_agent()) as resp: + while chunk := resp.read(_PROXY_CHUNK_SIZE): + yield chunk + except Exception as e: + log.warning("Proxy stream ended: user=%s source=%s: %s", username, source_id, e) + finally: + release_stream(username, source_id) diff --git a/m3u_export_test.py b/m3u_export_test.py new file mode 100644 index 0000000..6d23a15 --- /dev/null +++ b/m3u_export_test.py @@ -0,0 +1,392 @@ +"""Tests for m3u_export.py and the /get.php, /xmltv.php, /live/ routes.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import json + +import pytest + +import cache as cache_module + + +XTREAM_STREAM = { + "stream_id": 42, + "name": "News HD", + "stream_icon": "http://logo/news.png", + "epg_channel_id": "news.example", + "category_ids": ["src1_5"], + "source_id": "src1", + "source_type": "xtream", + "source_url": "http://provider.example", + "source_username": "puser", + "source_password": "ppass", +} + +M3U_STREAM = { + "stream_id": "src2_1", + "name": "Local OTA", + "stream_icon": "", + "epg_channel_id": "ota.example", + "category_ids": ["src2_ota"], + "direct_url": "http://192.168.1.87:5004/auto/v2.1", + "source_id": "src2", +} + +CATEGORIES = [ + {"category_id": "src1_5", "category_name": "News", "source_id": "src1"}, + {"category_id": "src2_ota", "category_name": "OTA", "source_id": "src2"}, +] + + +@pytest.fixture +def export_module(tmp_path: Path): + """Import m3u_export with temp auth settings and a clean state.""" + import auth + import m3u_export + + with ( + patch.object(auth, "SERVER_SETTINGS_FILE", tmp_path / "server_settings.json"), + patch.object(auth, "USERS_DIR", tmp_path / "users"), + ): + (tmp_path / "users").mkdir(exist_ok=True) + m3u_export.clear_credential_cache() + m3u_export._active_streams.clear() + yield m3u_export + + +class TestVerifyCredentials: + def test_valid_and_cached(self, export_module): + import auth + + auth.create_user("alice", "password123") + assert export_module.verify_credentials("alice", "password123") + # Second call must hit the cache, not PBKDF2 + with patch.object(auth, "verify_password", side_effect=AssertionError) as _: + assert export_module.verify_credentials("alice", "password123") + + def test_invalid_password(self, export_module): + import auth + + auth.create_user("alice", "password123") + assert not export_module.verify_credentials("alice", "wrong") + + def test_wrong_password_not_served_from_cache(self, export_module): + import auth + + auth.create_user("alice", "password123") + assert export_module.verify_credentials("alice", "password123") + assert not export_module.verify_credentials("alice", "different") + + def test_clear_cache_forces_reverify(self, export_module): + import auth + + auth.create_user("alice", "password123") + assert export_module.verify_credentials("alice", "password123") + auth.change_password("alice", "newpassword1") + export_module.clear_credential_cache("alice") + assert not export_module.verify_credentials("alice", "password123") + assert export_module.verify_credentials("alice", "newpassword1") + + +class TestStreamFiltering: + def test_allowed_when_no_restrictions(self, export_module): + with patch("auth.get_user_limits", return_value={"unavailable_groups": []}): + streams = export_module.allowed_live_streams([XTREAM_STREAM, M3U_STREAM], "alice") + assert len(streams) == 2 + + def test_blocked_category_filtered(self, export_module): + with patch( + "auth.get_user_limits", return_value={"unavailable_groups": ["cat:src1_5"]} + ): + streams = export_module.allowed_live_streams([XTREAM_STREAM, M3U_STREAM], "alice") + assert [s["name"] for s in streams] == ["Local OTA"] + + def test_stream_allowed_rule(self, export_module): + assert not export_module.stream_allowed(XTREAM_STREAM, {"cat:src1_5"}) + assert export_module.stream_allowed(XTREAM_STREAM, {"cat:other"}) + + +class TestUpstreamUrl: + def test_m3u_direct_url(self, export_module): + assert export_module.upstream_url(M3U_STREAM) == "http://192.168.1.87:5004/auto/v2.1" + + def test_xtream_ts_url(self, export_module): + assert ( + export_module.upstream_url(XTREAM_STREAM) + == "http://provider.example/live/puser/ppass/42.ts" + ) + + def test_xtream_credentials_urlencoded(self, export_module): + stream = {**XTREAM_STREAM, "source_password": "p#ss"} + assert "p%23ss" in export_module.upstream_url(stream) + + def test_unknown_stream(self, export_module): + assert export_module.upstream_url({"stream_id": 1}) == "" + + +class TestStripStreamExt: + def test_strips_known_extensions(self, export_module): + assert export_module.strip_stream_ext("42.ts") == "42" + assert export_module.strip_stream_ext("42.m3u8") == "42" + assert export_module.strip_stream_ext("src2_1.ts") == "src2_1" + + def test_keeps_other_ids(self, export_module): + assert export_module.strip_stream_ext("42") == "42" + assert export_module.strip_stream_ext("src2_1") == "src2_1" + + +class TestBuildPlaylist: + def test_playlist_format(self, export_module): + playlist = export_module.build_playlist( + CATEGORIES, [XTREAM_STREAM, M3U_STREAM], "https://tv.example", "alice", "pw" + ) + lines = playlist.strip().split("\n") + assert lines[0] == "#EXTM3U" + assert 'tvg-id="news.example"' in lines[1] + assert 'group-title="News"' in lines[1] + assert lines[1].endswith(",News HD") + assert lines[2] == "https://tv.example/live/alice/pw/42.ts" + assert 'group-title="OTA"' in lines[3] + assert lines[4] == "https://tv.example/live/alice/pw/src2_1.ts" + + def test_credentials_urlencoded_in_urls(self, export_module): + playlist = export_module.build_playlist( + CATEGORIES, [XTREAM_STREAM], "https://tv.example", "alice", "p w#1" + ) + assert "/live/alice/p%20w%231/42.ts" in playlist + + +class TestBuildXmltv: + def test_channels_and_programs(self, export_module, tmp_path): + from datetime import UTC, datetime, timedelta + + import epg + + epg.init(tmp_path) + now = datetime.now(UTC) + epg.insert_programs( + [ + ( + "news.example", + "Evening News", + now.timestamp(), + (now + timedelta(hours=1)).timestamp(), + "Daily news & ", + "src1", + ) + ] + ) + epg.commit() + + xml = export_module.build_xmltv([XTREAM_STREAM, M3U_STREAM]) + assert '' in xml + assert "News HD" in xml + assert '' in xml + assert "Evening News" in xml + assert "Daily news & <updates>" in xml + + def test_streams_without_epg_id_skipped(self, export_module, tmp_path): + import epg + + epg.init(tmp_path) + xml = export_module.build_xmltv([{**M3U_STREAM, "epg_channel_id": ""}]) + assert " str: + """Base URL as seen by the client (honors reverse proxy headers).""" + scheme = request.headers.get("x-forwarded-proto") or request.url.scheme + host = ( + request.headers.get("x-forwarded-host") + or request.headers.get("host") + or request.url.netloc + ) + return f"{scheme}://{host}" + + +def _require_export_auth(request: Request, username: str, password: str) -> None: + """Auth for export endpoints (credentials in URL, not cookies).""" + if not load_server_settings().get("m3u_export_enabled"): + raise HTTPException(404, "Not Found") + ip = request.client.host if request.client else "unknown" + _check_rate_limit(ip) + if not m3u_export.verify_credentials(username, password): + _login_attempts.setdefault(ip, []).append(time.time()) + raise HTTPException(401, "Invalid credentials") + + +def _load_live_cats_and_streams() -> tuple[list[dict], list[dict]]: + """Load live categories and streams into cache (without EPG fetch).""" + if "live_streams" not in get_cache(): + cats, streams, epg_urls = load_all_live_data() + with get_cache_lock(): + get_cache()["live_categories"] = cats + get_cache()["live_streams"] = streams + get_cache()["epg_urls"] = epg_urls + return get_cache().get("live_categories", []), get_cache().get("live_streams", []) + + +@app.get("/get.php") +async def m3u_export_playlist(request: Request, username: str = "", password: str = ""): + """Xtream-style M3U playlist of the user's allowed live channels.""" + _require_export_auth(request, username, password) + cats, streams = await asyncio.to_thread(_load_live_cats_and_streams) + streams = m3u_export.allowed_live_streams(streams, username) + content = m3u_export.build_playlist( + cats, streams, _external_base_url(request), username, password + ) + return Response( + content=content, + media_type="audio/x-mpegurl", + headers={"Content-Disposition": 'attachment; filename="playlist.m3u"'}, + ) + + +@app.get("/xmltv.php") +async def m3u_export_xmltv(request: Request, username: str = "", password: str = ""): + """XMLTV guide for the user's allowed live channels.""" + _require_export_auth(request, username, password) + _, streams = await asyncio.to_thread(_load_live_cats_and_streams) + streams = m3u_export.allowed_live_streams(streams, username) + content = await asyncio.to_thread(m3u_export.build_xmltv, streams) + return Response(content=content, media_type="application/xml") + + +@app.get("/live/{username}/{password}/{stream_spec}") +async def m3u_export_live(request: Request, username: str, password: str, stream_spec: str): + """Play a live channel from the exported playlist (redirect or proxy).""" + _require_export_auth(request, username, password) + stream_id = m3u_export.strip_stream_ext(stream_spec) + _, streams = await asyncio.to_thread(_load_live_cats_and_streams) + stream = next((s for s in streams if str(s.get("stream_id")) == stream_id), None) + if not stream: + raise HTTPException(404, "Channel not found") + user_limits = auth.get_user_limits(username) + if not m3u_export.stream_allowed(stream, set(user_limits.get("unavailable_groups", []))): + raise HTTPException(403, "Access to this channel is restricted") + url = m3u_export.upstream_url(stream) + if not url: + raise HTTPException(404, "Stream not found") + mode = load_server_settings().get("m3u_export_mode", "redirect") + # HLS direct URLs are always redirected: proxying them would require + # rewriting the segment URLs inside the playlist. + if mode != "proxy" or url.endswith(".m3u8"): + return RedirectResponse(url, status_code=302) + source_id = stream.get("source_id", "") + if not m3u_export.try_acquire_stream(username, source_id): + raise HTTPException(429, "Stream limit reached") + return StreamingResponse( + m3u_export.iter_proxy_stream(url, username, source_id), + media_type="video/mp2t", + ) + + # ============================================================================= # Transcoding routes (logic in ffmpeg_session.py) # ============================================================================= @@ -2120,6 +2214,9 @@ async def settings_page(request: Request, user: Annotated[dict, Depends(require_ "probe_series": server_settings.get("probe_series", False), "user_agent_preset": server_settings.get("user_agent_preset", "default"), "user_agent_custom": server_settings.get("user_agent_custom", ""), + "m3u_export_enabled": server_settings.get("m3u_export_enabled", False), + "m3u_export_mode": server_settings.get("m3u_export_mode", "redirect"), + "external_base_url": _external_base_url(request), "available_encoders": AVAILABLE_ENCODERS, "sr_available": is_sr_available(), "sr_models": get_sr_models(), @@ -2675,6 +2772,21 @@ async def settings_transcode( return {"ok": True} +@app.post("/settings/m3u-export") +async def settings_m3u_export( + _user: Annotated[dict, Depends(require_admin)], + m3u_export_enabled: Annotated[str | None, Form()] = None, + m3u_export_mode: Annotated[str, Form()] = "redirect", +): + settings = load_server_settings() + settings["m3u_export_enabled"] = m3u_export_enabled == "on" + settings["m3u_export_mode"] = ( + m3u_export_mode if m3u_export_mode in ("redirect", "proxy") else "redirect" + ) + save_server_settings(settings) + return {"ok": True} + + @app.post("/settings/refresh-encoders") async def settings_refresh_encoders( _user: Annotated[dict, Depends(require_admin)], @@ -2804,6 +2916,8 @@ async def update_settings_api( "probe_series", "vod_order", "series_order", + "m3u_export_enabled", + "m3u_export_mode", } settings = load_server_settings() for key in allowed_keys: @@ -2855,6 +2969,7 @@ async def settings_delete_user( if not password or not auth.verify_password(username, password): raise HTTPException(400, "Password required to delete your own account") auth.delete_user(username) + m3u_export.clear_credential_cache(username) response = RedirectResponse("/login", status_code=303) response.delete_cookie("token") return response @@ -2863,6 +2978,7 @@ async def settings_delete_user( raise HTTPException(403, "Admin access required") if not auth.delete_user(username): raise HTTPException(404, "User not found") + m3u_export.clear_credential_cache(username) return RedirectResponse("/settings", status_code=303) @@ -2912,6 +3028,7 @@ async def settings_change_own_password( raise HTTPException(400, "Password must be at least 8 characters") if not auth.change_password(username, new_password): raise HTTPException(404, "User not found") + m3u_export.clear_credential_cache(username) return {"status": "ok"} @@ -2929,6 +3046,7 @@ async def settings_change_password( raise HTTPException(400, "Password must be at least 8 characters") if not auth.change_password(target_user, new_password): raise HTTPException(404, "User not found") + m3u_export.clear_credential_cache(target_user) return {"status": "ok"} diff --git a/static/js/settings.js b/static/js/settings.js index 7ac743b..148b937 100644 --- a/static/js/settings.js +++ b/static/js/settings.js @@ -575,6 +575,26 @@ customInput?.addEventListener('change', function() { save(this); }); } + function setupM3uExportSettings() { + const container = document.getElementById('m3u-export-settings'); + if (!container) return; + + const enabledCb = container.querySelector('input[name="m3u_export_enabled"]'); + const urlsBox = document.getElementById('m3u-export-urls'); + + async function save(triggerEl) { + const form = new FormData(); + if (enabledCb?.checked) form.append('m3u_export_enabled', 'on'); + form.append('m3u_export_mode', container.querySelector('input[name="m3u_export_mode"]:checked')?.value || 'redirect'); + await saveWithFeedback('/settings/m3u-export', { method: 'POST', body: form }, getFeedbackEl(triggerEl)); + urlsBox?.classList.toggle('hidden', !enabledCb?.checked); + } + + container.querySelectorAll('.setting-input').forEach(el => { + el.addEventListener('change', function() { save(this); }); + }); + } + // ============================================================ // Data & Probe Cache // ============================================================ @@ -926,6 +946,7 @@ setupGuideSettings(); setupTranscodeSettings(); setupUserAgentSettings(); + setupM3uExportSettings(); setupDataCache(); setupProbeCache(); setupRefreshButtons(); diff --git a/templates/settings.html b/templates/settings.html index 6b779a6..4c9c290 100644 --- a/templates/settings.html +++ b/templates/settings.html @@ -653,6 +653,43 @@

User-Agent

+ +
+

M3U Export

+

Expose live channels as an M3U playlist + XMLTV guide for external IPTV players (TiviMate, IPTV Smarters, VLC). Users sign in with their neTV username and password; channel restrictions apply.

+
+ +
+ +
+ + +
+

Redirect: players fetch streams directly from the provider (no server bandwidth, provider URL visible to clients). Proxy: streams flow through this server (hides provider, enforces stream limits, uses your upload bandwidth).

+
+
+ +
+
Playlist: {{ external_base_url }}/get.php?username=USER&password=PASS
+
EPG: {{ external_base_url }}/xmltv.php?username=USER&password=PASS
+
+

Replace USER/PASS with the neTV account credentials.

+
+
+
+

Data Cache