From 45227be6bfd240530d3649034bf2ba0a46538e35 Mon Sep 17 00:00:00 2001
From: ArabCoders <admin@arabcoders.org>
Date: Tue, 17 Dec 2024 17:25:16 +0300
Subject: [PATCH] added LFI protection for thumbnails

---
 app/library/HttpAPI.py | 21 ++++++++++++-------
 app/library/Utils.py   | 47 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 60 insertions(+), 8 deletions(-)

diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py
index 034bb0d8..f1a8d2a9 100644
--- a/app/library/HttpAPI.py
+++ b/app/library/HttpAPI.py
@@ -4,7 +4,6 @@ import functools
 import json
 import os
 import time
-
 import httpx
 from .config import Config
 from .DownloadQueue import DownloadQueue
@@ -20,6 +19,7 @@ from .common import common
 from pathlib import Path
 from .encoder import Encoder
 from .Emitter import Emitter
+from .Utils import validate_url
 
 LOG = logging.getLogger('app')
 MIME = magic.Magic(mime=True)
@@ -527,6 +527,11 @@ class HttpAPI(common):
         if not url:
             return web.json_response({"error": "URL is required."}, status=400)
 
+        try:
+            validate_url(url)
+        except Exception as e:
+            return web.json_response({"error": str(e)}, status=400)
+
         try:
             opts = {
                 'proxy': self.config.ytdl_options.get('proxy', None),
@@ -539,13 +544,13 @@ class HttpAPI(common):
             async with httpx.AsyncClient(**opts) as client:
                 LOG.info(f"Fetching thumbnail from '{url}'.")
                 response = await client.request(method='GET', url=url)
-                return web.Response(body=response.content, headers={
-                    'Content-Type': response.headers.get('Content-Type'),
-                    'Pragma': 'public',
-                    'Access-Control-Allow-Origin': '*',
-                    'Cache-Control': f"public, max-age={time.time() + 31536000}",
-                    'Expires': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(time.time() + 31536000).timetuple()),
-                })
+                return web.Response(body=response.content,
+                                    headers={'Content-Type': response.headers.get('Content-Type'),
+                                             'Pragma': 'public', 'Access-Control-Allow-Origin': '*',
+                                             'Cache-Control': f"public, max-age={time.time() + 31536000}",
+                                             'Expires': time.strftime(
+                                                 '%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(
+                                                     time.time() + 31536000).timetuple()), })
         except Exception as e:
             LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'")
             return web.json_response({"error": str(e)}, status=500)
diff --git a/app/library/Utils.py b/app/library/Utils.py
index e7b41755..bb87df3c 100644
--- a/app/library/Utils.py
+++ b/app/library/Utils.py
@@ -1,10 +1,13 @@
 import copy
 from datetime import datetime, timezone
+from functools import lru_cache
+import ipaddress
 import json
 import logging
 import os
 import pathlib
 import re
+import socket
 from typing import Any
 import uuid
 import yt_dlp
@@ -359,3 +362,47 @@ def ag(array: dict | list, path: list[str | int] | str | int, default: Any = Non
             return get_value(default)
 
     return current
+
+
+@lru_cache(maxsize=512)
+def is_private_address(hostname: str) -> bool:
+    try:
+        ip = socket.gethostbyname(hostname)
+        ip_obj = ipaddress.ip_address(ip)
+        return (ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local)
+    except socket.gaierror:
+        # Could not resolve - treat as invalid or restricted
+        return True
+
+
+def validate_url(url: str) -> bool:
+    """
+    Validate if the url is valid and allowed.
+
+    Args:
+        url (str): URL to validate.
+
+    Raises:
+        ValueError: If the URL is invalid or not allowed.
+
+    Returns:
+        bool: True if the URL is valid and allowed.
+    """
+    if not url:
+        raise ValueError("URL is required.")
+
+    try:
+        from yarl import URL
+        parsed_url = URL(url)
+    except ValueError:
+        raise ValueError("Invalid URL.")
+
+    # Check allowed schemes
+    if parsed_url.scheme not in ["http", "https"]:
+        raise ValueError("Invalid scheme usage. Only HTTP or HTTPS allowed.")
+
+    hostname = parsed_url.host
+    if not hostname or is_private_address(hostname):
+        raise ValueError("Access to internal urls or private networks is not allowed.")
+
+    return True