diff --git a/src/updater.py b/src/updater.py index 50973c4053d..ff336b7eaeb 100644 --- a/src/updater.py +++ b/src/updater.py @@ -109,6 +109,8 @@ TOP_CONTRIBUTORS_FILENAME = f'{TOP_CONTRIBUTORS_BASENAME}.svg' JSON_EXTENSION = '.json' PNG_CONTENT_TYPE = 'image/png' +DATABASE_ITEM_ID_PATTERN = re.compile(r'\d+') +IMDB_ID_PATTERN = re.compile(r'tt\d+') APPROVE_THEME_LABEL = 'approve-theme' ISSUE_AUTHOR_BADGE_STYLE = 'for-the-badge' ISSUE_AUTHOR_BADGES = { @@ -1310,14 +1312,37 @@ def _update_issue_audit_data(og_data: dict, _write_auto_close_message(message=DUPLICATE_NO_REPLACEMENT_REASON_CLOSE_MESSAGE) +def _build_database_json_path(base_dir: str, item_id: Union[int, str], pattern: re.Pattern, label: str) -> str: + """Build a database JSON path from an expected identifier format.""" + item_id_text = str(item_id) + if not pattern.fullmatch(item_id_text): + raise ValueError(f'Invalid {label}: {item_id_text}') + + return os.path.join(os.path.abspath(base_dir), f'{item_id_text}{JSON_EXTENSION}') + + def _write_item_files(database_path: str, item_type: str, og_data: dict) -> None: """Write the updated database item files.""" - destination_filenames = [os.path.join(database_path, f'{og_data["id"]}.json')] + destination_filenames = [ + _build_database_json_path( + base_dir=database_path, + item_id=og_data['id'], + pattern=DATABASE_ITEM_ID_PATTERN, + label='database item id', + ), + ] if item_type == 'movie': try: if og_data["imdb_id"]: - destination_filenames.append(os.path.join(imdb_path, f'{og_data["imdb_id"]}.json')) + destination_filenames.append( + _build_database_json_path( + base_dir=imdb_path, + item_id=og_data['imdb_id'], + pattern=IMDB_ID_PATTERN, + label='IMDb id', + ), + ) except KeyError as e: print_github_error(f'Error getting imdb_id: {e}') diff --git a/tests/unit/test_issue_updater.py b/tests/unit/test_issue_updater.py index 5b5a2083cf5..b0a7d64dba5 100644 --- a/tests/unit/test_issue_updater.py +++ b/tests/unit/test_issue_updater.py @@ -499,6 +499,34 @@ def test_write_item_files_writes_primary_and_imdb_copy(tmp_path, monkeypatch): assert json.loads((imdb_dir / 'tt0113189.json').read_text()) == item_data +@pytest.mark.parametrize('unsafe_id', ['../710', '710/evil', '710.json']) +def test_write_item_files_rejects_unsafe_database_id(tmp_path, unsafe_id): + """Test database item file writing rejects unsafe item identifiers.""" + database_path = tmp_path / 'database' / 'movies' / 'themoviedb' + item_data = {'id': unsafe_id, 'title': 'GoldenEye'} + + with pytest.raises(ValueError, match='Invalid database item id'): + updater._write_item_files(database_path=str(database_path), item_type='movie_collection', og_data=item_data) + + assert not database_path.exists() + assert not (tmp_path / 'database' / 'movies' / '710.json').exists() + + +@pytest.mark.parametrize('unsafe_imdb_id', ['../tt0113189', 'tt0113189/evil', 'nm0113189']) +def test_write_item_files_rejects_unsafe_imdb_id(tmp_path, monkeypatch, unsafe_imdb_id): + """Test movie file writing rejects unsafe IMDb identifiers before writing files.""" + database_path = tmp_path / 'database' / 'movies' / 'themoviedb' + imdb_dir = tmp_path / 'database' / 'movies' / 'imdb' + item_data = {'id': 710, 'imdb_id': unsafe_imdb_id, 'title': 'GoldenEye'} + monkeypatch.setattr(updater, 'imdb_path', str(imdb_dir)) + + with pytest.raises(ValueError, match='Invalid IMDb id'): + updater._write_item_files(database_path=str(database_path), item_type='movie', og_data=item_data) + + assert not database_path.exists() + assert not imdb_dir.exists() + + def test_load_issue_submission_values_uses_supplied_values(monkeypatch): """Test fully supplied issue-update values bypass submission loading but still validate YouTube.""" monkeypatch.setattr(