Enhance downloader and parser modules with improved error handling and anti-hotlinking protection

- Added common file magic bytes for detecting anti-hotlink responses in downloader.
- Updated content validation to return detailed error reasons for invalid TS content.
- Enhanced decryption logic to log strategies and results for better debugging.
- Improved M3U8 parser with logging of playlist content previews and segment counts.
- Introduced additional headers in download requests to bypass anti-hotlinking measures.
- Implemented checks for anti-hotlinking errors during segment downloads, with appropriate logging and error handling.

Author: asdfghj1237890

m3u8-downloader/docker/worker/downloader.py

@@ -22,6 +22,11 @@
 TS_SYNC_BYTE = b'\x47'
 TS_PACKET_SIZE = 188
 
+# Common file magic bytes for detecting anti-hotlink responses
+JPEG_MAGIC = b'\xff\xd8\xff'
+PNG_MAGIC = b'\x89PNG'
+GIF_MAGIC = b'GIF8'
+
 
 class SegmentDownloader:
     """Download video segments with multi-threading and retry logic"""
@@ -56,22 +61,30 @@ def __init__(
         # Create output directory
         self.output_dir.mkdir(parents=True, exist_ok=True)
 
-    def _is_valid_ts_content(self, data: bytes) -> bool:
+    def _is_valid_ts_content(self, data: bytes) -> tuple[bool, str]:
         """
         Validate if the content is a valid MPEG-TS file.
-        Returns False if it looks like HTML/text error page.
+        Returns (is_valid, error_reason) tuple.
         """
         if not data or len(data) < TS_PACKET_SIZE:
-            return False
+            return False, "Content too small"
+        
+        # Check for image files (anti-hotlinking protection)
+        if data[:3] == JPEG_MAGIC:
+            return False, "Server returned JPEG image (anti-hotlinking protection)"
+        if data[:4] == PNG_MAGIC:
+            return False, "Server returned PNG image (anti-hotlinking protection)"
+        if data[:4] == GIF_MAGIC:
+            return False, "Server returned GIF image (anti-hotlinking protection)"
 
         # Check if it starts with HTML (error page)
         if data[:5].lower() in (b'<!doc', b'<html', b'<?xml'):
-            return False
+            return False, "Server returned HTML error page"
 
         # Check for common error text patterns
         lower_start = data[:500].lower()
         if b'error' in lower_start or b'forbidden' in lower_start or b'denied' in lower_start:
-            return False
+            return False, "Server returned error response"
 
         # Check for TS sync byte at expected positions
         # TS packets are 188 bytes, sync byte should appear at 0, 188, 376, etc.
@@ -81,32 +94,62 @@ def _is_valid_ts_content(self, data: bytes) -> bool:
                 sync_count += 1
 
         # If we found sync bytes at expected positions, it's likely valid
-        return sync_count >= 2
+        if sync_count >= 2:
+            return True, ""
+        
+        return False, "Invalid TS format (no sync bytes found)"
 
     def _decrypt_segment(self, data: bytes, segment_index: int) -> bytes:
         """Decrypt AES-128 encrypted segment"""
         if not self.encryption_key:
             return data
 
-        try:
-            # Use provided IV or derive from segment index
-            if self.encryption_iv:
-                iv = self.encryption_iv
+        # Log key info on first segment
+        if segment_index == 0:
+            logger.info(f"Encryption key (first 4 bytes): {self.encryption_key[:4].hex()}")
+            if self.encryption_iv is not None:
+                logger.info(f"Using provided IV: {self.encryption_iv.hex()}")
             else:
-                # Default IV is segment sequence number as 16-byte big-endian
-                iv = segment_index.to_bytes(16, byteorder='big')
+                logger.info("No IV provided, will use segment index")
+        
+        try:
+            # Try multiple IV strategies
+            iv_strategies = []
 
-            cipher = AES.new(self.encryption_key, AES.MODE_CBC, iv)
-            decrypted = cipher.decrypt(data)
+            # Strategy 1: Use provided IV if specified (HLS spec compliant)
+            if self.encryption_iv is not None:
+                iv_strategies.append(("provided IV", self.encryption_iv))
 
-            # Remove PKCS7 padding
-            try:
-                decrypted = unpad(decrypted, AES.block_size)
-            except ValueError:
-                # Some streams don't use proper padding
-                pass
+            # Strategy 2: Use segment index as IV (common non-compliant streams)
+            iv_strategies.append(("segment index IV", segment_index.to_bytes(16, byteorder='big')))
+            
+            # Strategy 3: Use zeros IV if not already tried
+            if self.encryption_iv is None or self.encryption_iv != bytes(16):
+                iv_strategies.append(("zeros IV", bytes(16)))
+            
+            for strategy_name, iv in iv_strategies:
+                cipher = AES.new(self.encryption_key, AES.MODE_CBC, iv)
+                decrypted = cipher.decrypt(data)
+                
+                # Remove PKCS7 padding
+                try:
+                    decrypted = unpad(decrypted, AES.block_size)
+                except ValueError:
+                    # Some streams don't use proper padding
+                    pass
+                
+                # Check if decryption produced valid TS data
+                if decrypted[:1] == TS_SYNC_BYTE:
+                    if segment_index < 3:  # Log first few segments
+                        logger.info(f"Segment {segment_index}: Decryption successful with {strategy_name}")
+                    return decrypted
 
+            # None of the strategies worked
+            logger.warning(f"Segment {segment_index}: All decryption strategies failed (first byte after zeros IV: {hex(decrypted[0]) if decrypted else 'empty'})")
+            
+            # Return the last decrypted result (with zeros IV) - let ffmpeg try to handle it
             return decrypted
+            
         except Exception as e:
             logger.warning(f"Decryption failed for segment {segment_index}: {e}")
             return data  # Return original data if decryption fails
@@ -133,6 +176,11 @@ def download_segment(
         try:
             logger.debug(f"Downloading segment {index}: {url}")
 
+            # Log headers for first segment to help debug anti-hotlink issues
+            if index == 0 and retry_count == 0:
+                logger.info(f"Segment download headers: {self.headers}")
+                logger.info(f"First segment URL: {url}")
+            
             response = self.session.get(
                 url,
                 headers=self.headers,
@@ -161,12 +209,23 @@ def download_segment(
                 content = self._decrypt_segment(content, index)
 
             # Validate content is actually a TS file (not an error page)
-            if not self._is_valid_ts_content(content):
-                # Log first 200 bytes for debugging
-                preview = content[:200]
-                logger.error(f"Segment {index} content is not valid TS data")
-                logger.error(f"Content preview (first 200 bytes): {preview}")
-                raise ValueError(f"Invalid TS content - possibly HTML error page or encrypted data")
+            is_valid, error_reason = self._is_valid_ts_content(content)
+            if not is_valid:
+                # Check if this looks like encrypted data that we couldn't decrypt
+                # In that case, still save it and let ffmpeg try to handle it
+                skip_validation = os.environ.get('SKIP_TS_VALIDATION', 'false').lower() == 'true'
+                
+                # Always skip validation for encrypted streams where decryption produced non-image data
+                if self.encryption_key and not content[:3] in (JPEG_MAGIC, PNG_MAGIC[:3], GIF_MAGIC[:3]):
+                    logger.warning(f"Segment {index}: {error_reason} - saving anyway for ffmpeg to process")
+                elif skip_validation:
+                    logger.warning(f"Segment {index}: {error_reason} - validation skipped")
+                else:
+                    # Log first 200 bytes for debugging
+                    preview = content[:200]
+                    logger.error(f"Segment {index}: {error_reason}")
+                    logger.error(f"Content preview (first 200 bytes): {preview}")
+                    raise ValueError(error_reason)
 
             # Write validated content to file
             with open(output_path, 'wb') as f:

m3u8-downloader/docker/worker/m3u8_parser.py

@@ -64,22 +64,28 @@ def parse(self) -> Dict:
             # Fetch playlist content
             content = self.fetch_playlist()
 
+            # Log first 500 chars of content to diagnose parsing issues
+            content_preview = content[:500] if len(content) > 500 else content
+            logger.info(f"Playlist content preview ({len(content)} bytes):\n{content_preview}")
+            
             # Parse with m3u8 library
             playlist = m3u8.loads(content, uri=self.url)
 
             # Check if this is a master playlist (with variants)
             if playlist.is_variant:
                 logger.info("Master playlist detected, selecting best quality")
-                return self._parse_master_playlist(playlist)
+                return self._parse_master_playlist(playlist, content)
             else:
                 logger.info("Media playlist detected")
-                return self._parse_media_playlist(playlist)
+                # Debug: log segment count before parsing
+                logger.debug(f"Raw playlist has {len(playlist.segments)} segments, {len(playlist.playlists)} playlists")
+                return self._parse_media_playlist(playlist, content)
 
         except Exception as e:
             logger.error(f"Failed to parse m3u8: {e}")
             raise
 
-    def _parse_master_playlist(self, playlist: m3u8.M3U8) -> Dict:
+    def _parse_master_playlist(self, playlist: m3u8.M3U8, content: str = None) -> Dict:
         """Parse master playlist and select best quality variant"""
         if not playlist.playlists:
             raise ValueError("No variants found in master playlist")
@@ -108,13 +114,13 @@ def _parse_master_playlist(self, playlist: m3u8.M3U8) -> Dict:
         variant_content = variant_parser.fetch_playlist()
         variant_playlist = m3u8.loads(variant_content, uri=variant_url)
 
-        result = self._parse_media_playlist(variant_playlist)
+        result = self._parse_media_playlist(variant_playlist, variant_content)
         result['resolution'] = resolution
         result['selected_variant_url'] = variant_url
 
         return result
 
-    def _parse_media_playlist(self, playlist: m3u8.M3U8) -> Dict:
+    def _parse_media_playlist(self, playlist: m3u8.M3U8, content: str = None) -> Dict:
         """Parse media playlist and extract segment URLs"""
         segments = []
         total_duration = 0.0
@@ -132,6 +138,10 @@ def _parse_media_playlist(self, playlist: m3u8.M3U8) -> Dict:
             total_duration += segment.duration
 
         if not segments:
+            # Log the actual content for debugging
+            if content:
+                content_preview = content[:1000] if len(content) > 1000 else content
+                logger.error(f"Playlist content (no segments found):\n{content_preview}")
             raise ValueError("No segments found in playlist")
 
         logger.info(f"Found {len(segments)} segments, total duration: {total_duration:.1f}s")
@@ -165,15 +175,27 @@ def _get_encryption_info(self, playlist: m3u8.M3U8) -> Optional[Dict]:
                     response.raise_for_status()
                     key = response.content
 
+                    # Validate key length (AES-128 requires 16 bytes)
+                    logger.info(f"Encryption key length: {len(key)} bytes")
+                    if len(key) != 16:
+                        logger.warning(f"Unexpected key length: {len(key)} bytes (expected 16)")
+                        # Some servers return key with extra whitespace or headers
+                        if len(key) > 16:
+                            logger.info(f"Key preview (first 32 bytes): {key[:32]}")
+                    
                     # Get IV from key info or use default
                     iv = None
                     if segment.key.iv:
                         # IV is usually specified as hex string like 0x...
                         iv_str = segment.key.iv
+                        logger.info(f"IV from m3u8: {iv_str}")
                         if iv_str.startswith('0x') or iv_str.startswith('0X'):
                             iv = bytes.fromhex(iv_str[2:])
                         else:
                             iv = bytes.fromhex(iv_str)
+                        logger.info(f"Parsed IV length: {len(iv)} bytes, value: {iv.hex()}")
+                    else:
+                        logger.info("No IV specified in m3u8, will use segment sequence number")
 
                     return {
                         'method': 'AES-128',

m3u8-downloader/docker/worker/worker.py

@@ -318,6 +318,20 @@ def _process_m3u8_download(self, job_id: str, job: dict):
             if 'User-Agent' not in headers:
                 headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36'
 
+            # Add additional browser-like headers to bypass anti-hotlinking
+            if 'Accept' not in headers:
+                headers['Accept'] = '*/*'
+            if 'Accept-Language' not in headers:
+                headers['Accept-Language'] = 'en-US,en;q=0.9'
+            if 'Accept-Encoding' not in headers:
+                headers['Accept-Encoding'] = 'gzip, deflate, br'
+            if 'Sec-Fetch-Dest' not in headers:
+                headers['Sec-Fetch-Dest'] = 'empty'
+            if 'Sec-Fetch-Mode' not in headers:
+                headers['Sec-Fetch-Mode'] = 'cors'
+            if 'Sec-Fetch-Site' not in headers:
+                headers['Sec-Fetch-Site'] = 'cross-site'
+            
             # Debug: Log headers to verify
             logger.info(f"Request headers: {headers}")
             if 'Cookie' in headers:
@@ -351,10 +365,22 @@ def _process_m3u8_download(self, job_id: str, job: dict):
             logger.info("Step 2: Downloading segments")
             temp_dir = tempfile.mkdtemp(prefix=f"m3u8_{job_id}_")
 
+            # Create segment-specific headers - some CDNs expect Referer to be the m3u8 URL
+            segment_headers = headers.copy()
+            m3u8_base_url = playlist_info.get('base_url', job['url'])
+            parsed_m3u8 = urlparse(m3u8_base_url)
+            m3u8_origin = f"{parsed_m3u8.scheme}://{parsed_m3u8.netloc}"
+            
+            # Use m3u8 URL as Referer for segments (common anti-hotlink bypass)
+            segment_headers['Referer'] = m3u8_base_url
+            segment_headers['Origin'] = m3u8_origin
+            logger.info(f"Segment Referer set to: {m3u8_base_url}")
+            logger.info(f"Segment Origin set to: {m3u8_origin}")
+            
             downloader = SegmentDownloader(
                 segments=playlist_info['segments'],
                 output_dir=temp_dir,
-                headers=headers,
+                headers=segment_headers,
                 max_workers=int(os.getenv('MAX_DOWNLOAD_WORKERS', 2)),
                 encryption_key=playlist_info.get('encryption_key'),
                 encryption_iv=playlist_info.get('encryption_iv')
@@ -370,9 +396,19 @@ def progress_callback(completed, total):
                 download_progress = int(5 + (completed / total) * 80)
                 self.update_job_status(job_id, "downloading", progress=download_progress)
 
-                # Check if too many segments failed with 403/474 errors during download
+                # Check if too many segments failed during download
                 failed_count = len(downloader.failed_segments)
-                if failed_count > 20:
+                if failed_count > 5:
+                    # Count anti-hotlink protection errors
+                    hotlink_count = sum(
+                        1 for item in downloader.failed_segments 
+                        if 'anti-hotlinking' in item['error'].lower() or 'JPEG' in item['error'] or 'PNG' in item['error']
+                    )
+                    
+                    if hotlink_count >= 5:
+                        logger.error(f"Anti-hotlinking protection detected: {hotlink_count} segments blocked")
+                        raise Exception(f"Download aborted: Server blocked segment downloads (anti-hotlinking protection). Try refreshing the source page and retrying.")
+                    
                     # Count HTTP 403/474 errors
                     http_error_count = sum(
                         1 for item in downloader.failed_segments