Fix #2750: Handle invalid LINK frames when saving MP3 files

MP3 files with LINK frames that do not follow the ID3 specification
cause mutagen exceptions when saving with ID3 v2.3 format. This fix
handles invalid LINK frames to prevent errors during save operations.

Author: francescouteiro

picard/formats/id3.py

@@ -170,6 +170,7 @@ class ID3File(File):
         'TSOT': 'titlesort',
         'WCOP': 'license',
         'WOAR': 'website',
+        'WXXX': 'user_website',
         'COMM': 'comment',
         'TOAL': 'originalalbum',
         'TOPE': 'originalartist',
@@ -300,6 +301,8 @@ def _load(self, filename):
         for frame in tags.values():
             self._process_frame(frame, metadata, config_params)
 
+        self._process_link_frames_on_load(tags, metadata)
+
         if 'date' in metadata:
             self._sanitize_date(metadata)
 
@@ -614,6 +617,12 @@ def _get_tags(self, filename):
 
     def _save_tags(self, tags, filename):
         config = get_config()
+
+        try:
+            self._sanitize_id3_frames(tags)
+        except Exception as e:
+            log.error("Error sanitizing ID3 frames: %s", e)
+
         if config.setting['write_id3v1']:
             v1 = 2
         else:
@@ -622,11 +631,153 @@ def _save_tags(self, tags, filename):
         if config.setting['write_id3v23']:
             tags.update_to_v23()
             separator = config.setting['id3v23_join_with']
-            tags.save(filename, v2_version=3, v1=v1, v23_sep=separator)
+            try:
+                tags.save(filename, v2_version=3, v1=v1, v23_sep=separator)
+            except ValueError as e:
+                if str(e) == "Invalid frame ID":
+                    log.warning("Invalid frame ID error when saving. Removing all LINK frames as fallback...")
+                    for frame_id in list(tags.keys()):
+                        if frame_id.startswith('LINK'):
+                            del tags[frame_id]
+                    tags.save(filename, v2_version=3, v1=v1, v23_sep=separator)
+                else:
+                    raise
         else:
             tags.update_to_v24()
             tags.save(filename, v2_version=4, v1=v1)
 
+    def _sanitize_id3_frames(self, tags):
+        """This method attempts to fix various issues with ID3 frames:
+        1. Handle invalid LINK frames (convert to WXXX if possible)
+        2. Remove null bytes and other illegal characters from frame IDs
+        Note: In the future, we might consider prompting users about invalid frames."""
+        to_remove = []
+        to_add = []
+        confirmed_problematic_frames = []
+        extracted_urls = []
+        for frame_id, frame in list(tags.items()):
+            try: 
+                invalid_id = False
+                if '\x00' in frame_id:
+                    invalid_id = True
+                if frame_id.startswith('LINK'):
+                    is_malformed = False
+                    if ':' in frame_id and any(term in frame_id for term in ('http', 'www.', '://')):
+                        is_malformed = True
+
+                    if hasattr(frame, 'data') and frame.data:
+                        frame_data = frame.data.decode('latin1', errors='ignore')
+                        # A valid frame should have: URL text string, null byte, then optional data
+                        has_url_pattern = any(term in frame_data for term in ('http', 'www.', '://'))
+                        if frame_data.startswith('\x00') and any(term in frame_data[1:] for term in ('http', 'www.', '://')):
+                            is_malformed = True
+                        elif has_url_pattern and '\x00' not in frame_data:
+                            is_malformed = True
+                        elif '\x00' in frame_data:
+                            parts = frame_data.split('\x00', 1)
+                            if len(parts) > 1 and any(term in parts[1] for term in ('http', 'www.', '://')):
+                                is_malformed = True
+
+                    if is_malformed:
+                        invalid_id = True
+                        if self._extract_and_convert_link_frame(frame, frame_id, to_add):
+                            log.debug("Successfully converted malformed LINK frame %r to WXXX", frame_id)
+                            for item in to_add:
+                                if isinstance(item, mutagen.id3.WXXX):
+                                    extracted_urls.append(item.url)
+
+                if invalid_id:
+                    confirmed_problematic_frames.append(frame_id)
+                    to_remove.append(frame_id)
+                    
+            except Exception as e:
+                log.error("Error processing frame %r: %s", frame_id, e)
+
+        for frame_id in to_remove:
+            del tags[frame_id] 
+        for frame in to_add:
+            tags.add(frame)
+
+        if extracted_urls and hasattr(self, 'metadata') and hasattr(self, 'orig_metadata'):
+            if 'user_website' not in self.metadata:
+                self.metadata['user_website'] = extracted_urls
+            if 'user_website' not in self.orig_metadata:
+                self.orig_metadata['user_website'] = extracted_urls
+
+        return confirmed_problematic_frames
+
+    def _extract_and_convert_link_frame(self, frame, frame_id, to_add):
+        """Extract URL from a malformed LINK frame and convert it to a WXXX frame. Handles cases where URLs
+        are split between frame ID and data or separated by null bytes (common for archive.org)"""
+
+        try:
+            url_parts = []  
+            if hasattr(frame, 'data') and frame.data:
+                frame_data = frame.data.decode('latin1', errors='ignore')
+                if '\x00' in frame_data:
+                    parts = frame_data.split('\x00', 1)
+                    # Likely valid LINK frame if the first part looks like a URL and second part doesn't contain URL patterns
+                    if (parts[0] and any(term in parts[0] for term in ('http', 'www.', '://')) 
+                            and len(parts) > 1 and not any(term in parts[1] for term in ('http', 'www.', '://'))):
+                        log.debug("Skipping valid LINK frame with URL %r", parts[0])
+                        return False
+
+            if ':' in frame_id:
+                frame_id_parts = frame_id.split(':', 1)
+                if frame_id_parts[0] == 'LINK':
+                    # Standard case where everything after LINK: is part of the URL
+                    url_parts.append(frame_id_parts[1])
+                elif any(proto in frame_id_parts[0] for proto in ('LINKhttp', 'LINKhtt', 'LINKwww')):
+                    # Case where part of the URL protocol is in the frame_id
+                    prefix = 'LINK'
+                    protocol_part = frame_id_parts[0][len(prefix):]
+                    url_parts.append(protocol_part + ':' + frame_id_parts[1])
+
+            if hasattr(frame, 'data') and frame.data:
+                frame_data = frame.data.decode('latin1', errors='ignore').strip()
+                if frame_data:
+                    if '\x00' in frame_data:
+                        parts = frame_data.split('\x00')
+                        for part in parts:
+                            if part.strip() and any(term in part for term in ('http', 'www.', '://')):
+                                url_parts.append(part.strip())
+                    elif any(term in frame_data for term in ('http', 'www.', '://')):
+                        url_parts.append(frame_data)
+
+            combined_url = ''.join(url_parts).strip(':;,. \t\n\r')
+
+            if combined_url:
+                url_match = re.search(r'(?:https?:?(?:/+|\\+)|www\.)[a-zA-Z0-9][-a-zA-Z0-9\.]+\.[a-zA-Z]{2,}(?:/[^\s:;,]*)?', combined_url)
+                if url_match:
+                    raw_url = url_match.group(0)
+
+                    if raw_url.startswith('www.'):
+                        final_url = 'http://' + raw_url
+                    elif raw_url.startswith(('http://', 'https://')):
+                        final_url = raw_url
+                    else:
+                        final_url = re.sub(r'^htt:?p', 'http', raw_url)
+                        if not final_url.startswith(('http://', 'https://')):
+                            if final_url.startswith(':'):
+                                final_url = 'http' + final_url
+                            elif final_url.startswith('//'):
+                                final_url = 'http:' + final_url
+                            elif final_url.startswith('p://'):
+                                final_url = 'http://' + final_url[4:]
+                            else:
+                                final_url = 'http://' + final_url
+
+                    wxxx_frame = mutagen.id3.WXXX(encoding=Id3Encoding.LATIN1, desc="URL from malformed LINK frame", url=final_url)
+                    to_add.append(wxxx_frame)
+                    return True
+
+            log.warning("Could not extract URL from malformed LINK frame %r", frame_id)
+            return False
+
+        except Exception as e:
+            log.error("Failed to process LINK frame %r: %s", frame_id, e)
+            return False
+
     def format_specific_metadata(self, metadata, tag, settings=None):
         if not settings:
             settings = get_config().setting
@@ -1035,6 +1186,49 @@ def _remove_other_supported_tag(self, tags, real_name):
         """Remove other supported tag from ID3 frames."""
         del tags[real_name]
 
+    def _process_link_frames_on_load(self, tags, metadata):
+        """Process malformed LINK frames during initial file load and add them as website URLs."""
+
+        link_frames = []
+        for frame_id, frame in list(tags.items()):
+            if frame_id.startswith('LINK'):
+                is_potentially_malformed = False
+
+                if ':' in frame_id and any(term in frame_id for term in ('http', 'www.', '://')):
+                    is_potentially_malformed = True
+
+                if hasattr(frame, 'data') and frame.data:
+                    try:
+                        frame_data = frame.data.decode('latin1', errors='ignore')
+
+                        if frame_data.startswith('\x00') and any(term in frame_data[1:] for term in ('http', 'www.', '://')):
+                            is_potentially_malformed = True
+
+                        elif '\x00' in frame_data:
+                            parts = frame_data.split('\x00', 1)
+                            if len(parts) > 1 and any(term in parts[1] for term in ('http', 'www.', '://')):
+                                is_potentially_malformed = True
+
+                        elif any(term in frame_data for term in ('http', 'www.', '://')) and '\x00' not in frame_data:
+                            is_potentially_malformed = True
+                    except Exception:
+                        is_potentially_malformed = True
+
+                if is_potentially_malformed:
+                    link_frames.append((frame_id, frame))
+        
+        if not link_frames:
+            return
+
+        to_add = []
+        for frame_id, frame in link_frames:
+            if self._extract_and_convert_link_frame(frame, frame_id, to_add):
+                log.debug("Extracted URL from malformed LINK frame %r during load", frame_id)
+
+        for frame in to_add:
+            if isinstance(frame, mutagen.id3.WXXX):
+                url = frame.url
+                metadata.add('user_website', url)
 
 class MP3File(ID3File):
 
@@ -1080,10 +1274,26 @@ def _get_tags(self, filename):
 
     def _save_tags(self, tags, filename):
         config = get_config()
+
+        try:
+            self._sanitize_id3_frames(tags)
+        except Exception as e:
+            log.error("Error sanitizing ID3 frames: %s", e)
+
         if config.setting['write_id3v23']:
             compatid3.update_to_v23(tags)
             separator = config.setting['id3v23_join_with']
-            tags.save(filename, v2_version=3, v23_sep=separator)
+            try:
+                tags.save(filename, v2_version=3, v23_sep=separator)
+            except ValueError as e:
+                if str(e) == "Invalid frame ID":
+                    log.debug("Invalid frame ID error when saving. Removing all LINK frames as fallback...")
+                    for frame_id in list(tags.keys()):
+                        if frame_id.startswith('LINK'):
+                            del tags[frame_id]
+                    tags.save(filename, v2_version=3, v23_sep=separator)
+                else:
+                    raise
         else:
             tags.update_to_v24()
             tags.save(filename, v2_version=4)

picard/util/tags.py

@@ -136,6 +136,7 @@
     'totaltracks': N_('Total Tracks'),
     'tracknumber': N_('Track Number'),
     'website': N_('Artist Website'),
+    'user_website': N_('User Defined Website'),
     'work': N_('Work'),
     'writer': N_('Writer'),
 }

test/formats/test_id3.py

@@ -625,6 +625,70 @@ def test_releasedate_v24(self):
             raw_metadata = load_raw(self.filename)
             self.assertEqual(metadata['releasedate'], raw_metadata['TDRL'])
 
+        @skipUnlessTestfile
+        def test_malformed_link_url_in_data(self):
+            filename = self.copy_file_tmp(os.path.join('test', 'data', 'test.mp3'), '.mp3')
+            tags = mutagen.id3.ID3()
+            try:
+                tags.load(filename)
+            except mutagen.id3.ID3NoHeaderError:
+                pass
+            frame = mutagen.id3.Frames['LINK'](data=b'\x00http://example.org/null')
+            tags.add(frame)
+            tags.save(filename)
+            metadata = load_metadata(filename)
+            self.assertIn('user_website', metadata)
+            website_url = metadata['user_website']
+            self.assertTrue('http://example.org/null' in website_url)
+            save_metadata(filename, metadata)
+            raw_metadata = load_raw(filename)
+            has_url_frame = False
+            for key in raw_metadata:
+                if key.startswith('WXXX:') or key == 'WXXX':
+                    has_url_frame = True
+                    self.assertTrue(website_url in key or website_url == raw_metadata[key])
+                    break
+            self.assertTrue(has_url_frame, "No URL frame (WXXX) found in raw metadata")
+            self.assertFalse(any(k.startswith('LINK') for k in raw_metadata))
+
+        @skipUnlessTestfile
+        def test_archive_org_link_pattern(self):
+            filename = self.copy_file_tmp(os.path.join('test', 'data', 'test.mp3'), '.mp3')
+
+            tags = mutagen.id3.ID3()
+            try:
+                tags.load(filename)
+            except mutagen.id3.ID3NoHeaderError:
+                pass
+
+            frame = mutagen.id3.Frames['LINK'](data=b'p://www.archive.org/details/test_item')
+            tags.add(frame)
+            tags.save(filename)
+            
+            metadata = load_metadata(filename)
+            self.assertIn('user_website', metadata, "URL was not extracted from archive.org pattern")
+            website_url = metadata['user_website']
+            self.assertIn('archive.org', website_url, "URL doesn't contain archive.org domain")
+            self.assertIn('details/test_item', website_url, "URL path was not correctly extracted")
+            
+            save_metadata(filename, metadata)
+            raw_metadata = load_raw(filename)
+            has_url_frame = False
+            
+            for key in raw_metadata:
+                if key.startswith('WXXX:') or key == 'WXXX':
+                    has_url_frame = True
+                    frame = raw_metadata[key]
+                    if hasattr(frame, 'url'):
+                        self.assertIn('archive.org', frame.url)
+                        self.assertIn('details/test_item', frame.url)
+                    else:
+                        self.assertIn('archive.org', frame)
+                        self.assertIn('details/test_item', frame)
+                    break
+                
+            self.assertTrue(has_url_frame, "No URL frame (WXXX) found - LINK frame wasn't converted")
+            self.assertFalse(any(k.startswith('LINK') for k in raw_metadata), "LINK frame was not removed after conversion")
 
 class MP3Test(CommonId3Tests.Id3TestCase):
     testfile = 'test.mp3'