import json import mimetypes import re from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse from .config import Settings, load_settings from .database import Database from .security import create_token, 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}$") class TuTVApplication: def __init__(self, settings: Settings | None = None): self.settings = settings or load_settings() self.database = Database(self.settings.database_path) 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_DELETE(self): application.handle(self) def log_message(self, format, *args): return return Handler def handle(self, request: BaseHTTPRequestHandler): 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") if request.command == "GET" and path.startswith("/activaciones/"): relative = path.removeprefix("/activaciones/") return self._serve_file(request, self.settings.portal_dir / relative) if request.command == "POST" and path == "/api/v1/auth/login": return self._login(request) 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/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]) 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: return self._json( request, HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Error interno del servidor"}, ) def _body(self, request: BaseHTTPRequestHandler) -> dict: 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 _admin(self, request: BaseHTTPRequestHandler) -> str | None: authorization = request.headers.get("Authorization", "") if not authorization.startswith("Bearer "): return None subject = verify_token( authorization.removeprefix("Bearer ").strip(), self.settings.token_secret, ) return subject if subject == self.settings.admin_email else None def _login(self, request): body = self._body(request) if ( body.get("email") != self.settings.admin_email or body.get("password") != self.settings.admin_password ): return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "Acceso incorrecto"}) token = create_token(self.settings.admin_email, self.settings.token_secret) return self._json(request, HTTPStatus.OK, {"token": token}) def _register_device(self, request): body = self._body(request) code = str(body.get("code", "")).upper() 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): code = request.headers.get("X-Device-Code", "").upper() 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 self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "Dispositivo inválido"}) if device["status"] != "active": return self._json(request, HTTPStatus.OK, {"status": "pending", "code": code}) return self._json( request, HTTPStatus.OK, { "status": "active", "deviceName": device["name"], "catalog": json.loads(device["catalog_json"] or "{}"), }, ) def _list_devices(self, request): admin = self._admin(request) if not admin: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) return self._json(request, HTTPStatus.OK, {"devices": self.database.list_devices(admin)}) def _activate_device(self, request): admin = self._admin(request) if not admin: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) body = self._body(request) code = str(body.get("code", "")).upper() name = str(body.get("name", "")).strip() playlist_url = str(body.get("playlistUrl", "")).strip() if not DEVICE_CODE.fullmatch(code) or not name: return self._json(request, HTTPStatus.BAD_REQUEST, {"error": "Complete código y nombre"}) if not self.database.get_device_by_code(code): return self._json( request, HTTPStatus.NOT_FOUND, {"error": "Ese código aún no ha sido generado por una app TuTV"}, ) try: credentials = parse_xtream_url(playlist_url) 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, admin, name, encrypt_text(source, self.settings.encryption_key), catalog, ) 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: str): admin = self._admin(request) if not admin: return self._json(request, HTTPStatus.UNAUTHORIZED, {"error": "No autorizado"}) deleted = self.database.delete_device(code.upper(), admin) return self._json( request, HTTPStatus.OK if deleted else HTTPStatus.NOT_FOUND, {"deleted": deleted}, ) def _serve_file(self, request, path: Path): portal = self.settings.portal_dir.resolve() try: resolved = path.resolve() resolved.relative_to(portal) 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(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: HTTPStatus, payload: dict): 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) def serve(host: str = "127.0.0.1", port: int = 8080): app = TuTVApplication() server = ThreadingHTTPServer((host, port), app.handler_class()) print(f"TuTV disponible en http://{host}:{port}/activaciones") server.serve_forever()