import json import mimetypes import re import traceback from http import HTTPStatus from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from socketserver import ThreadingMixIn from urllib.parse import parse_qs, unquote, urlparse from .config import Settings, load_settings from .database import Database from .security import create_token, decrypt_text, encrypt_text, hash_secret, verify_token from .xtream import XtreamClient, XtreamError, parse_xtream_url, pending_catalog DEVICE_CODE = re.compile(r"^[A-Z0-9]{2}(?::[A-Z0-9]{2}){4}$") def normalize_device_code(value): compact = re.sub(r"[^A-Z0-9]", "", str(value or "").upper()) if len(compact) != 10: return str(value or "").upper() return ":".join(compact[index : index + 2] for index in range(0, 10, 2)) class TuTVApplication: def __init__(self, settings=None): self.settings = settings or load_settings() self.database = Database(self.settings.database_path) self.database.ensure_admin(self.settings.admin_email, hash_secret(self.settings.admin_password)) def handler_class(self): application = self class Handler(BaseHTTPRequestHandler): server_version = "TuTV/1.0" def do_GET(self): application.handle(self) def do_POST(self): application.handle(self) def do_PATCH(self): application.handle(self) def do_DELETE(self): application.handle(self) def log_message(self, format, *args): return return Handler def handle(self, request): path = urlparse(request.path).path.rstrip("/") or "/" try: if request.command == "GET" and path in {"/", "/activaciones"}: return self._serve_file(request, self.settings.portal_dir / "index.html", self.settings.portal_dir) if request.command == "GET" and path.startswith("/activaciones/"): return self._serve_file(request, self.settings.portal_dir / path[len("/activaciones/") :], self.settings.portal_dir) if request.command == "GET" and path == "/updates/version.json": return self._serve_file(request, self.settings.updates_dir / "version.json", self.settings.updates_dir) if request.command == "GET" and path.startswith("/updates/"): return self._serve_file(request, self.settings.updates_dir / path[len("/updates/") :], self.settings.updates_dir) if request.command == "POST" and path == "/api/v1/auth/login": return self._login(request) if request.command == "GET" and path == "/api/v1/admin/me": return self._me(request) if request.command == "PATCH" and path == "/api/v1/admin/me": return self._update_me(request) if request.command == "GET" and path == "/api/v1/admin/users": return self._list_users(request) if request.command == "POST" and path == "/api/v1/admin/users": return self._create_user(request) if request.command == "PATCH" and path.startswith("/api/v1/admin/users/"): return self._update_user(request, path.rsplit("/", 1)[-1]) if request.command == "DELETE" and path.startswith("/api/v1/admin/users/"): return self._delete_user(request, path.rsplit("/", 1)[-1]) if request.command == "GET" and path == "/api/v1/admin/devices": return self._list_devices(request) if request.command == "POST" and path == "/api/v1/admin/devices/activate": return self._activate_device(request) if request.command == "DELETE" and path.startswith("/api/v1/admin/devices/"): return self._delete_device(request, path.rsplit("/", 1)[-1]) if request.command == "POST" and path == "/api/v1/device/register": return self._register_device(request) if request.command == "GET" and path == "/api/v1/device/catalog": return self._device_catalog(request) if request.command == "GET" and path == "/api/v1/device/items": return self._device_items(request) if request.command == "GET" and path == "/api/v1/device/series/seasons": return self._series_seasons(request) if request.command == "GET" and path == "/api/v1/device/series/episodes": return self._series_episodes(request) if request.command == "GET" and path == "/api/v1/app/version": return self._serve_file(request, self.settings.updates_dir / "version.json", self.settings.updates_dir) if request.command == "GET" and path == "/api/v1/app/latest.apk": return self._serve_file(request, self.settings.updates_dir / "tutv-latest.apk", self.settings.updates_dir) return self._json(request, HTTPStatus.NOT_FOUND, {"error": "Ruta no encontrada"}) except (ValueError, json.JSONDecodeError): return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Solicitud inválida"}) except Exception: traceback.print_exc() return self._json(request, HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Error interno del servidor"}) def _body(self, request): length = int(request.headers.get("Content-Length", "0")) if length > 1_000_000: raise ValueError("body too large") return json.loads(request.rfile.read(length) or b"{}") def _current_user(self, request): authorization = request.headers.get("Authorization", "") if not authorization.startswith("Bearer "): return None subject = verify_token(authorization[len("Bearer ") :].strip(), self.settings.token_secret) user = self.database.get_user(subject) if subject else None return user if user and user.get("active") else None def _login(self, request): body = self._body(request) email = str(body.get("email", "")).strip().lower() password = str(body.get("password", "")) user = self.database.get_user(email) if not user or not user.get("active") or user.get("password_hash") != hash_secret(password): return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "Acceso incorrecto"}) token = create_token(user["email"], self.settings.token_secret) return self._json(request, HTTPStatus.OK, {"token": token, "user": self._public_user(user, user)}) def _me(self, request): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) return self._json(request, HTTPStatus.OK, {"user": self._public_user(user, user)}) def _list_users(self, request): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) if user["role"] == "reseller": return self._json(request, HTTPStatus.FORBIDDEN, {"error": "No puede ver usuarios"}) users = [self._public_user(item, user) for item in self.database.list_users(user)] return self._json(request, HTTPStatus.OK, {"users": users}) def _create_user(self, request): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) body = self._body(request) email = str(body.get("email", "")).strip().lower() password = str(body.get("password", "")) role = str(body.get("role", "")).strip() host = self._clean_host(str(body.get("xtreamHost", "")).strip()) if user["role"] == "administrator" else (user.get("xtream_host") or "") if not email or "@" not in email or len(password) < 6: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Correo o contraseña inválidos"}) if not self._can_create_role(user, role): return self._json(request, HTTPStatus.FORBIDDEN, {"error": "No puede crear ese rol"}) if role in {"supervisor", "reseller"} and not host: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Asigne un host Xtream"}) created = self.database.create_user(email, hash_secret(password), role, user["email"], host) return self._json(request, HTTPStatus.OK, {"user": self._public_user(created, user)}) def _update_user(self, request, email): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) target = self.database.get_user(unquote(email).lower()) if not target or not self._can_manage_user(user, target): return self._json(request, HTTPStatus.NOT_FOUND, {"error": "Usuario no encontrado"}) body = self._body(request) new_email = str(body.get("email", "")).strip().lower() password = str(body.get("password", "")) host = body.get("xtreamHost") active = body.get("active") host_update = self._clean_host(str(host).strip()) if host is not None and user["role"] == "administrator" else None updated = self.database.update_user( target["email"], new_email if new_email and user["role"] == "administrator" else None, hash_secret(password) if password else None, host_update, bool(active) if active is not None else None, ) payload = {"user": self._public_user(updated, user)} if target["email"] == user["email"] and updated["email"] != user["email"]: payload["token"] = create_token(updated["email"], self.settings.token_secret) return self._json(request, HTTPStatus.OK, payload) def _update_me(self, request): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) body = self._body(request) password = str(body.get("password", "")) host = body.get("xtreamHost") if password and len(password) < 6: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "La clave debe tener 6 caracteres o más"}) updated = self.database.update_user( user["email"], None, hash_secret(password) if password else None, self._clean_host(str(host).strip()) if host is not None and user["role"] == "administrator" else None, None, ) return self._json(request, HTTPStatus.OK, {"user": self._public_user(updated, updated)}) def _delete_user(self, request, email): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) target = self.database.get_user(unquote(email).lower()) if not target or not self._can_delete_user(user, target): return self._json(request, HTTPStatus.NOT_FOUND, {"error": "Usuario no encontrado"}) result = self.database.delete_user_tree(target["email"]) return self._json(request, HTTPStatus.OK, result) def _register_device(self, request): body = self._body(request) code = normalize_device_code(body.get("code", "")) secret = str(body.get("deviceSecret", "")) if not DEVICE_CODE.fullmatch(code) or len(secret) < 32: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Código o secreto de dispositivo inválido"}) device = self.database.register_device(code, hash_secret(secret)) return self._json(request, HTTPStatus.OK, {"code": code, "status": device["status"]}) def _device_catalog(self, request): device = self._authorized_device(request) if not device: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "Dispositivo inválido"}) if device["status"] != "active": return self._json(request, HTTPStatus.OK, {"status": "pending", "code": request.headers.get("X-Device-Code", "").upper()}) return self._json( request, HTTPStatus.OK, {"status": "active", "deviceName": device["name"], "catalog": json.loads(device["catalog_json"] or "{}")}, ) def _device_items(self, request): client, error = self._device_xtream_client(request) if error: return error query = parse_qs(urlparse(request.path).query) group = query.get("group", [""])[0] category_id = query.get("categoryId", [""])[0] if group not in {"live", "movies", "series"} or not category_id: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Categoría inválida"}) return self._json(request, HTTPStatus.OK, {"items": client.list_items(group, category_id)}) def _series_seasons(self, request): client, error = self._device_xtream_client(request) if error: return error series_id = parse_qs(urlparse(request.path).query).get("seriesId", [""])[0] if not series_id: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Serie inválida"}) return self._json(request, HTTPStatus.OK, {"seasons": client.list_series_seasons(series_id)}) def _series_episodes(self, request): client, error = self._device_xtream_client(request) if error: return error query = parse_qs(urlparse(request.path).query) series_id = query.get("seriesId", [""])[0] season_id = query.get("seasonId", [""])[0] if not series_id or not season_id: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Temporada inválida"}) return self._json(request, HTTPStatus.OK, {"episodes": client.list_series_episodes(series_id, season_id)}) def _device_xtream_client(self, request): device = self._authorized_device(request) if not device: return None, self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "Dispositivo inválido"}) if device["status"] != "active": return None, self._json(request, HTTPStatus.CONFLICT, {"error": "Dispositivo pendiente"}) source = json.loads(decrypt_text(device["source_encrypted"], self.settings.encryption_key)) credentials = parse_xtream_url( "{}/get.php?username={}&password={}".format(source["baseUrl"], source["username"], source["password"]) ) return XtreamClient(credentials), None def _authorized_device(self, request): code = normalize_device_code(request.headers.get("X-Device-Code", "")) secret = request.headers.get("X-Device-Secret", "") device = self.database.get_device_by_code(code) if not device or device["device_secret_hash"] != hash_secret(secret): return None return device def _list_devices(self, request): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) search = parse_qs(urlparse(request.path).query).get("search", [""])[0].strip() return self._json(request, HTTPStatus.OK, {"devices": self.database.list_devices(user, search)}) def _activate_device(self, request): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) body = self._body(request) code = normalize_device_code(body.get("code", "")) name = str(body.get("name", "")).strip() if not DEVICE_CODE.fullmatch(code) or not name: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Complete código y nombre"}) device = self.database.get_device_by_code(code) if not device: return self._json(request, HTTPStatus.NOT_FOUND, {"error": "Ese código aún no ha sido generado por una app TuTV"}) if device.get("owner_email") and not self.database.can_manage_owner(user, device.get("owner_email")): return self._json(request, HTTPStatus.FORBIDDEN, {"error": "No puede editar ese dispositivo"}) try: credentials = self._credentials_from_body(user, body) except XtreamError as exc: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": str(exc)}) catalog = pending_catalog() sync_warning = None if body.get("syncNow", True): try: catalog = XtreamClient(credentials).build_catalog() except XtreamError as exc: sync_warning = str(exc) source = json.dumps({"baseUrl": credentials.base_url, "username": credentials.username, "password": credentials.password}) self.database.activate( code, device.get("owner_email") or user["email"], name, encrypt_text(source, self.settings.encryption_key), catalog, credentials.username, credentials.base_url, ) response = {"status": "active", "code": code} if sync_warning: response["warning"] = sync_warning return self._json(request, HTTPStatus.OK, response) def _delete_device(self, request, code): user = self._current_user(request) if not user: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) deleted = self.database.delete_device(normalize_device_code(unquote(code)), user) return self._json(request, HTTPStatus.OK if deleted else HTTPStatus.NOT_FOUND, {"deleted": deleted}) def _credentials_from_body(self, user, body): playlist_url = str(body.get("playlistUrl", "")).strip() username = str(body.get("username", "")).strip() password = str(body.get("password", "")).strip() if user["role"] == "administrator" and playlist_url: return parse_xtream_url(playlist_url) host = self._clean_host(str(body.get("xtreamHost", "")).strip()) if user["role"] == "administrator" else user.get("xtream_host") if not host: raise XtreamError("Ese usuario no tiene host Xtream asignado") if not username or not password: raise XtreamError("Complete usuario y clave Xtream") return parse_xtream_url("{}/get.php?username={}&password={}&type=m3u_plus&output=mpegts".format(host, username, password)) def _clean_host(self, value): value = value.strip() if not value: return "" if not value.startswith(("http://", "https://")): value = "http://" + value parsed = urlparse(value) if not parsed.scheme or not parsed.netloc: return "" return "{}://{}".format(parsed.scheme, parsed.netloc).rstrip("/") def _can_create_role(self, user, role): if user["role"] == "administrator": return role in {"supervisor", "reseller"} if user["role"] == "supervisor": return role == "reseller" return False def _can_manage_user(self, user, target): if user["role"] == "administrator": return True if user["role"] == "supervisor": return target["email"] == user["email"] or target.get("parent_email") == user["email"] return target["email"] == user["email"] def _can_delete_user(self, user, target): if user["email"] == target["email"]: return False if user["role"] == "administrator": return True if user["role"] == "supervisor": return target["role"] == "reseller" and target.get("parent_email") == user["email"] return False def _public_user(self, user, requester=None): show_host = bool(requester and requester.get("role") == "administrator") return { "email": user["email"], "role": user["role"], "parentEmail": user.get("parent_email"), "xtreamHost": (user.get("xtream_host") or "") if show_host else "", "active": bool(user.get("active")), "createdAt": user.get("created_at"), "updatedAt": user.get("updated_at"), } def _serve_file(self, request, path, base_dir=None): allowed_base = (base_dir or self.settings.portal_dir).resolve() try: resolved = path.resolve() resolved.relative_to(allowed_base) except (ValueError, FileNotFoundError): return self._json(request, HTTPStatus.NOT_FOUND, {"error": "Archivo no encontrado"}) if not resolved.is_file(): return self._json(request, HTTPStatus.NOT_FOUND, {"error": "Archivo no encontrado"}) data = resolved.read_bytes() request.send_response(HTTPStatus.OK) request.send_header("Content-Type", mimetypes.guess_type(str(resolved))[0] or "application/octet-stream") request.send_header("Content-Length", str(len(data))) request.end_headers() request.wfile.write(data) def _json(self, request, status, payload): data = json.dumps(payload, ensure_ascii=False).encode() request.send_response(status) request.send_header("Content-Type", "application/json; charset=utf-8") request.send_header("Cache-Control", "no-store") request.send_header("Content-Length", str(len(data))) request.end_headers() request.wfile.write(data) class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True def serve(host="127.0.0.1", port=8080): app = TuTVApplication() server = ThreadingHTTPServer((host, port), app.handler_class()) print("TuTV disponible en http://{}:{}/activaciones".format(host, port)) server.serve_forever()